ArgoWorkflow教程(四)---Workflow & 日志归档-数字星空

上一篇我们分析了argo-workflow 中的 artifact，包括 artifact-repository 配置以及 Workflow 中如何使用 artifact。本篇主要分析流水线 GC 以及归档,防止无限占用集群中 etcd 的空间。

1. 概述

因为 ArgoWorkflow 是用 CRD 方式实现的，不需要外部存储服务也可以正常运行：

运行记录使用 Workflow CR 对象存储
运行日志则存放在 Pod 中，通过 kubectl logs 方式查看
- 因此需要保证 Pod 不被删除，否则就无法查看了

但是也正因为所有数据都存放在集群中，当数据量大之后 etcd 存储压力会很大，最终影响到集群稳定性。

为了解决该问题 ArgoWorkflow 提供了归档功能，将历史数据归档到外部存储，以降低 etcd 的存储压力。

具体实现为：

1）将 Workflow 对象会存储到 Postgres(或 MySQL)
2）将 Pod 对应的日志会存储到 S3，因为日志数据量可能会比较大，因此没有直接存 PostgresQL。

为了提供归档功能，需要依赖两个存储服务：

Postgres：外部数据库，用于存储归档后的工作流记录
minio：提供 S3 存储，用于存储 Workflow 中生成的 artifact 以及已归档工作流的 Pod 日志

因此，如果不需要存储太多 Workflow 记录及日志查看需求的话，就不需要使用归档功能，定时清理集群中的数据即可。

2.Workflow GC

Argo Workflows 有个工作流执行记录（Workflow）的清理机制，也就是 Garbage Collect(GC)。GC 机制可以避免有太多的执行记录，防止 Kubernetes 的后端存储 Etcd 过载。

开启

我们可以在 ConfigMap 中配置期望保留的工作执行记录数量，这里支持为不同状态的执行记录设定不同的保留数量。

首先查看 argo-server 启动命令中指定的是哪个 Configmap

# kubectl -n argo get deploy argo-workflows-server -oyaml|grep args -A 5
      - args:
        - server
        - --configmap=argo-workflows-workflow-controller-configmap
        - --auth-mode=server
        - --secure=false
        - --loglevel

可以看到，这里是用的argo-workflows-workflow-controller-configmap，那么修改这个即可。

配置如下：

apiVersion: v1
data:
  retentionPolicy: |
    completed: 3
    failed: 3
    errored: 3
kind: ConfigMap
metadata:
  name: argo-workflows-workflow-controller-configmap
  namespace: argo

需要注意的是，这里的清理机制会将多余的 Workflow 资源从 Kubernetes 中删除。如果希望能更多历史记录的话，建议启用并配置好归档功能。

然后重启 argo-workflow-controller 和 argo-server

kubectl -n argo rollout restart deploy argo-workflows-server
kubectl -n argo rollout restart deploy argo-workflows-workflow-controller

测试

运行多个流水线，看下是否会自动清理

for ((i=1; i<=10; i++)); do
cat <<EOF | kubectl create -f -
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: hello-world-
spec:
  entrypoint: whalesay
  templates:
  - name: whalesay
    container:
      image: docker/whalesay
      command: [cowsay]
      args: ["hello world $i"]
EOF
done

创建了 10 个 Workflow，看一下运行完成后会不会自动清理掉

[root@lixd-argo archive]# k get wf
NAME                STATUS      AGE   MESSAGE
hello-world-6hgb2   Succeeded   74s
hello-world-6pl5w   Succeeded   37m
hello-world-9fdmv   Running     21s
hello-world-f464p   Running     18s
hello-world-kqwk4   Running     16s
hello-world-kxbtk   Running     18s
hello-world-p88vd   Running     19s
hello-world-q7xbk   Running     22s
hello-world-qvv7d   Succeeded   10m
hello-world-t94pb   Running     23s
hello-world-w79q6   Running     15s
hello-world-wl4vl   Running     23s
hello-world-znw7w   Running     23s

过一会再看

[root@lixd-argo archive]# k get wf
NAME                STATUS      AGE    MESSAGE
hello-world-f464p   Succeeded   102s
hello-world-kqwk4   Succeeded   100s
hello-world-w79q6   Succeeded   99s

可以看到，只保留了 3 条记录，其他的都被清理了，说明 GC 功能 ok。

3. 流水线归档

https://argo-workflows.readthedocs.io/en/stable/workflow-archive/

开启 GC 功能之后，会自动清理 Workflow 以保证 etcd 不被占满，但是也无法查询之前的记录了。

ArgoWorkflow 也提供了流水线归档功能，来解决该问题。

通过将 Workflow 记录到外部 Postgres 数据库来实现持久化，从而满足查询历史记录的需求。

部署 Postgres

首先，简单使用 helm 部署一个 AIO 的Postgres

REGISTRY_NAME=registry-1.docker.io
REPOSITORY_NAME=bitnamicharts
storageClass="local-path"
# postgres 账号的密码
adminPassword="postgresadmin"

helm install pg-aio oci://$REGISTRY_NAME/$REPOSITORY_NAME/postgresql \
--set global.storageClass=$storageClass \
--set global.postgresql.auth.postgresPassword=$adminPassword \
--set global.postgresql.auth.database=argo

配置流水线归档

同样的，在 argo 配置文件中增加 persistence 相关配置即可：

persistence: 
  archive: true
  postgresql:
    host: pg-aio-postgresql.default.svc.cluster.local
    port: 5432
    database: postgres
    tableName: argo_workflows
    userNameSecret:
      name: argo-postgres-config
      key: username
    passwordSecret:
      name: argo-postgres-config
      key: password

argo-workflows-workflow-controller-configmap 完整内容如下：

apiVersion: v1
data:
  retentionPolicy: |
    completed: 3
    failed: 3
    errored: 3
  persistence: |
    archive: true
    archiveTTL: 180d
    postgresql:
      host: pg-aio-postgresql.default.svc.cluster.local
      port: 5432
      database: argo
      tableName: argo_workflows
      userNameSecret:
        name: argo-postgres-config
        key: username
      passwordSecret:
        name: argo-postgres-config
        key: password
kind: ConfigMap
metadata:
  name: argo-workflows-workflow-controller-configmap
  namespace: argo

然后还要创建一个 secret

kubectl create secret generic argo-postgres-config -n argo --from-literal=password=postgresadmin --from-literal=username=postgres

可能还需要给 rbac，否则 Controller 无法查询 secret

kubectl create clusterrolebinding argo-workflow-controller-admin --clusterrole=admin --serviceaccount=argo:argo-workflows-workflow-controller

然后重启 argo-workflow-controller 和 argo-server

kubectl -n argo rollout restart deploy argo-workflows-server
kubectl -n argo rollout restart deploy argo-workflows-workflow-controller

在启用存档的情况下启动工作流控制器时，将在数据库中创建以下表：

argo_workflows
argo_archived_workflows
argo_archived_workflows_labels
schema_history

归档记录 GC

配置文件中的 archiveTTL 用于指定压缩到 Postgres 中的 Workflow 记录存活时间，argo Controller 会根据该配置自动删除到期的记录，若不指定该值则不会删除。

具体如下：

func (r *workflowArchive) DeleteExpiredWorkflows(ttl time.Duration) error {
	rs, err := r.session.SQL().
		DeleteFrom(archiveTableName).
		Where(r.clusterManagedNamespaceAndInstanceID()).
		And(fmt.Sprintf("finishedat < current_timestamp - interval '%d' second", int(ttl.Seconds()))).
		Exec()
	if err != nil {
		return err
	}
	rowsAffected, err := rs.RowsAffected()
	if err != nil {
		return err
	}
	log.WithFields(log.Fields{"rowsAffected": rowsAffected}).Info("Deleted archived workflows")
	return nil
}

不过删除任务默认每天执行一次，因此就算配置为 1m 分钟也不会立即删除。

func (wfc *WorkflowController) archivedWorkflowGarbageCollector(stopCh <-chan struct{}) {
	defer runtimeutil.HandleCrash(runtimeutil.PanicHandlers...)

	periodicity := env.LookupEnvDurationOr("ARCHIVED_WORKFLOW_GC_PERIOD", 24*time.Hour)
	if wfc.Config.Persistence == nil {
		log.Info("Persistence disabled - so archived workflow GC disabled - you must restart the controller if you enable this")
		return
	}
	if !wfc.Config.Persistence.Archive {
		log.Info("Archive disabled - so archived workflow GC disabled - you must restart the controller if you enable this")
		return
	}
	ttl := wfc.Config.Persistence.ArchiveTTL
	if ttl == config.TTL(0) {
		log.Info("Archived workflows TTL zero - so archived workflow GC disabled - you must restart the controller if you enable this")
		return
	}
	log.WithFields(log.Fields{"ttl": ttl, "periodicity": periodicity}).Info("Performing archived workflow GC")
	ticker := time.NewTicker(periodicity)
	defer ticker.Stop()
	for {
		select {
		case <-stopCh:
			return
		case <-ticker.C:
			log.Info("Performing archived workflow GC")
			err := wfc.wfArchive.DeleteExpiredWorkflows(time.Duration(ttl))
			if err != nil {
				log.WithField("err", err).Error("Failed to delete archived workflows")
			}
		}
	}
}

需要设置环境变量 ARCHIVED_WORKFLOW_GC_PERIOD 来调整该值，修改 argo-workflows-workflow-controller 增加 env，就像这样：

        env:
        - name: ARCHIVED_WORKFLOW_GC_PERIOD
          value: 1m

测试

接下来创建 Workflow 看下是否测试

for ((i=1; i<=10; i++)); do
cat <<EOF | kubectl create -f -
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: hello-world-
spec:
  entrypoint: whalesay
  templates:
  - name: whalesay
    container:
      image: docker/whalesay
      command: [cowsay]
      args: ["hello world $i"]
EOF
done

查看下是 postgres 中是否生成归档记录

export POSTGRES_PASSWORD=postgresadmin

kubectl run postgresql-dev-client --rm --tty -i --restart='Never' --namespace default --image docker.io/bitnami/postgresql:14.1.0-debian-10-r80 --env="PGPASSWORD=$POSTGRES_PASSWORD" --command -- psql --host pg-aio-postgresql -U postgres -d argo -p 5432

按 Enter 进入 Pod 后直接查询即可

# 查询表
argo-# \dt
                     List of relations
 Schema |              Name              | Type  |  Owner
--------+--------------------------------+-------+----------
 public | argo_archived_workflows        | table | postgres
 public | argo_archived_workflows_labels | table | postgres
 public | argo_workflows                 | table | postgres
 public | schema_history                 | table | postgres
(4 rows)

# 查询记录
argo=# select name,phase from argo_archived_workflows;
       name        |   phase
-------------------+-----------
 hello-world-s8v4f | Succeeded
 hello-world-6pl5w | Succeeded
 hello-world-qvv7d | Succeeded
 hello-world-vgjqr | Succeeded
 hello-world-g2s8f | Succeeded
 hello-world-jghdm | Succeeded
 hello-world-fxtvk | Succeeded
 hello-world-tlv9k | Succeeded
 hello-world-bxcg2 | Succeeded
 hello-world-f6mdw | Succeeded
 hello-world-dmvj6 | Succeeded
 hello-world-btknm | Succeeded
(12 rows)

# \q 退出
argo=# \q

可以看到，Postgres 中已经存储好了归档的 Workflow，这样需要查询历史记录时到 Postgres 查询即可。

将 archiveTTL 修改为 1 分钟，然后重启 argo，等待 1 至2 分钟后，再次查看

argo=#  select name,phase from argo_archived_workflows;
 name | phase
------+-------
(0 rows)

argo=#

可以看到，所有记录都因为 TTL 被清理了，这样也能保证外部 Postgres 中的数据不会越累积越多。

4. Pod 日志归档

https://argo-workflows.readthedocs.io/en/stable/configure-archive-logs/

流水线归档实现了流水线持久化，即使把集群中的 Workflow 对象删除了，也可以从 Postgres 中查询到记录以及状态等信息。

但是流水线执行的日志却分散在对应 Pod 中的，如果 Pod 被删除了，日志就无法查看了，因此我们还需要做日志归档。

配置 Pod 归档

全局配置

在 argo 配置文件中开启 Pod 日志归档并配置好 S3 信息。

具体配置如下：

和第三篇配置的 artifact 一样，只是多了一个 archiveLogs: true

artifactRepository:
  archiveLogs: true
  s3:
    endpoint: minio.default.svc:9000
    bucket: argo
    insecure: true
    accessKeySecret:
      name: my-s3-secret
      key: accessKey
    secretKeySecret:
      name: my-s3-secret
      key: secretKey

完整配置如下：

apiVersion: v1
data:
  retentionPolicy: |
    completed: 3
    failed: 3
    errored: 3
  persistence: |
    archive: true
    postgresql:
      host: pg-aio-postgresql.default.svc.cluster.local
      port: 5432
      database: argo
      tableName: argo_workflows
      userNameSecret:
        name: argo-postgres-config
        key: username
      passwordSecret:
        name: argo-postgres-config
        key: password
  artifactRepository: |
    archiveLogs: true
    s3:
      endpoint: minio.default.svc:9000
      bucket: argo
      insecure: true
      accessKeySecret:
        name: my-s3-secret
        key: accessKey
      secretKeySecret:
        name: my-s3-secret
        key: secretKey
kind: ConfigMap
metadata:
  name: argo-workflows-workflow-controller-configmap
  namespace: argo

注意：根据第三篇分析 artifact，argo 中关于 artifactRepository 的信息包括三种配置方式：

1）全局配置
2）命名空间默认配置
3）Workflow 中指定配置

这里是用的全局配置方式，如果 Namespace 级别或者 Workflow 级别也配置了 artifactRepository 并指定了不开启日志归档，那么也不会归档的。

然后重启 argo

kubectl -n argo rollout restart deploy argo-workflows-server
kubectl -n argo rollout restart deploy argo-workflows-workflow-controller

在 Workflow & template 中配置

配置整个工作流都需要归档

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: archive-location-
spec:
  archiveLogs: true
  entrypoint: whalesay
  templates:
  - name: whalesay
    container:
      image: docker/whalesay:latest
      command: [cowsay]
      args: ["hello world"]

配置工作流中的某一个 template 需要归档。

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: archive-location-
spec:
  entrypoint: whalesay
  templates:
  - name: whalesay
    container:
      image: docker/whalesay:latest
      command: [cowsay]
      args: ["hello world"]
    archiveLocation:
      archiveLogs: true

小结

3 个地方都可以配置是否归档，就还挺麻烦的，根据官方文档，各个配置优先级如下：

workflow-controller config (on) > workflow spec (on/off) > template (on/off)

Controller Config Map	Workflow Spec	Template	are we archiving logs?
true	true	true	true
true	true	false	true
true	false	true	true
true	false	false	true
false	true	true	true
false	true	false	false
false	false	true	true
false	false	false	false

对应的代码实现：

// IsArchiveLogs determines if container should archive logs
// priorities: controller(on) > template > workflow > controller(off)
func (woc *wfOperationCtx) IsArchiveLogs(tmpl *wfv1.Template) bool {
	archiveLogs := woc.artifactRepository.IsArchiveLogs()
	if !archiveLogs {
		if woc.execWf.Spec.ArchiveLogs != nil {
			archiveLogs = *woc.execWf.Spec.ArchiveLogs
		}
		if tmpl.ArchiveLocation != nil && tmpl.ArchiveLocation.ArchiveLogs != nil {
			archiveLogs = *tmpl.ArchiveLocation.ArchiveLogs
		}
	}
	return archiveLogs
}

建议配置全局的就行了。

测试

接下来创建 Workflow 看下是否测试

cat <<EOF | kubectl create -f -
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: hello-world-
spec:
  entrypoint: whalesay
  templates:
  - name: whalesay
    container:
      image: docker/whalesay
      command: [cowsay]
      args: ["hello world"]
EOF

等待 Workflow 运行完成

# k get po
NAME                     READY   STATUS      RESTARTS   AGE
hello-world-6pl5w        0/2     Completed   0          53s
# k get wf
NAME                STATUS      AGE   MESSAGE
hello-world-6pl5w   Succeeded   55s

到 S3 查看是否有日志归档文件

可以看到，在指定 bucket 里已经存储了一个日志文件，以$bucket/$workflowName/$stepName 格式命名。

正常一个 Workflow 都会有多个 Step，每一个 step 分一个目录存储

内容就是 Pod 日志，具体如下：

 _____________ 
< hello world >
 ------------- 
    \
     \
      \     
                    ##        .            
              ## ## ##       ==            
           ## ## ## ##      ===            
       /""""""""""""""""___/ ===        
  ~~~ {~~ ~~~~ ~~~ ~~~~ ~~ ~ /  ===- ~~~   
       \______ o          __/            
        \    \        __/             
          \____\______/