Kubernetes Activities

Eight activities for orchestrating a Kubernetes cluster: k8s.apply, k8s.get, k8s.list, k8s.delete, k8s.scale, k8s.logs, k8s.wait and k8s.exec.

They speak to the cluster's REST API directly — no kubectl binary and no kubeconfig on the worker. Every activity names its cluster inline, so one workflow can target several clusters, and resources are resolved through API discovery, which means custom resources work exactly like built-in ones.

Setup

Every activity takes a connection object. Build it once in context: and reuse it, rather than repeating it per step.

Field	Type	Required	Default	Description
api_server	str	yes	—	API server base URL. Must be https://
token_secret_key	str	one-of	—	Secret holding a bearer token
client_cert_secret_key	str	one-of	—	Secret holding a client certificate (PEM)
client_key_secret_key	str	one-of	—	Secret holding the client private key (PEM)
ca_cert	str	no	null	CA bundle as inline PEM
ca_cert_secret_key	str	no	null	Secret holding the CA bundle. Exclusive with ca_cert
insecure_skip_tls_verify	bool	no	false	Skip certificate verification
namespace	str	no	"default"	Namespace used when an activity does not name one

Exactly one authentication method is required: a bearer token, or both of the client certificate and key. An unauthenticated connection is not expressible.

context:
  k8s: {}

# ...
- transform:
    name: prepare
    output_data:
      - k8s:
          api_server: "https://my-cluster.example.com:6443"
          token_secret_key: "K8S_PROD_TOKEN"
          ca_cert_secret_key: "K8S_PROD_CA"
          namespace: "payments"

Store the credentials once with builtin.secret.upload.

ca_cert is the one field that may be inline

Tokens and client keys have no inline form — they must be named through a *_secret_key. A CA certificate is public material, so forcing it through the secret store buys nothing, and ca_cert accepts PEM directly. See Secrets.

Client certificates and CA bundles are written to the worker's filesystem

The Kubernetes client accepts TLS material only as file paths, so a connection using ca_cert/ca_cert_secret_key or client-certificate auth writes those PEMs into a private (0700) temporary directory for the lifetime of the connection, removed when the worker shuts down. A bearer token never touches disk — it lives only in memory. Deployments that cannot accept certificates on disk should use token auth against a publicly-trusted endpoint.

Connections are cached per worker and keyed on the resolved credentials, so rotating a secret transparently produces a new connection rather than pinning a revoked token.

---

Safety

k8s.apply, k8s.delete and k8s.exec can destroy a production cluster. Two independent mechanisms guard them.

1. Authorization policy. Every activity asserts a privilege on the internal.k8s_activities resource, with a per-verb action so a deployment can grant broad read-only access and narrow write access:

Action	Activities
read	k8s.get, k8s.list, k8s.logs, k8s.wait
write	k8s.apply, k8s.scale
delete	k8s.delete
exec	k8s.exec

The evidence offered to the policy includes api_server_host, namespace, kind, k8s_action, cluster_scoped and insecure_skip_tls_verify — so a policy can say things like "production workflows may only reach the production cluster" or "only these two wfspecs may exec". The policy itself is supplied by your deployment through the internal.authz_policies workflow, not by a file in this repository. See Authz Activities.

If no policy is deployed, the check passes

Like every internal.* privilege in Moco, an unresolvable policy is logged and allowed. Do not treat "k8s activities are authorization-gated" as protection until you have actually deployed a policy for internal.k8s_activities.

2. Hard guards, always on. These need no configuration and cannot be turned off except by an explicit per-call flag:

• k8s.delete refuses a cluster-scoped resource (Namespace, PersistentVolume, CRD, ClusterRole, …) unless allow_cluster_scoped: true. Scope comes from API discovery, so it is correct for custom resources too.
• k8s.delete refuses a label-selector delete unless allow_bulk_delete: true, and rejects an empty selector outright — label_selector: "" must never mean "everything".
• k8s.apply and k8s.delete refuse kube-system, kube-public and kube-node-lease unless allow_system_namespaces: true.
• k8s.apply rejects a document with no kind, no apiVersion, or no metadata.name/generateName, naming the offending document by index.
• k8s.exec takes a command list; there is no implicit shell.

---

k8s.apply

Applies one or more manifests using server-side apply, so the API server arbitrates field ownership and a conflict with another controller (an HPA owning replicas, Argo owning the spec) is a loud error rather than a silent overwrite.

A string manifest may contain several ----separated documents. They are applied in the order given — that is how you express Namespace-before-workload or CRD-before-custom-resource. There is no dependency sorting.

Input

Field	Type	Required	Default	Description
connection	object	yes	—	Cluster connection
manifest	str | object | list	yes	—	YAML text, one document, or a list of documents
namespace	str	no	connection's	Namespace for documents that do not name one
field_manager	str	no	"moco"	Server-side-apply field manager
force_conflicts	bool	no	false	Take ownership of fields another manager owns
dry_run	bool	no	false	Validate and admit without persisting
continue_on_error	bool	no	false	Apply every document, recording per-document errors
allow_system_namespaces	bool	no	false	Permit kube-system and friends

Output

Field	Type	Description
resources	list	One entry per document, in input order
resources[].api_version	str	apiVersion of the document
resources[].kind	str	Kind of the document
resources[].name	str | null	Name as returned by the API server
resources[].namespace	str | null	Namespace; null when cluster-scoped
resources[].uid	str | null	metadata.uid
resources[].resource_version	str | null	metadata.resourceVersion after the apply
resources[].error	str | null	Failure message; only set when continue_on_error is true
applied_count	int	Number of documents applied without error

By default the first failing document stops the apply and raises, naming the document's index and listing what had already been applied — because a half-mutated cluster is something the workflow needs to know about precisely.

- activity:
    type: k8s.apply
    name: create-job
    input_data:
      connection: "{{ k8s }}"
      field_manager: "moco-migrate"
      manifest:
        apiVersion: batch/v1
        kind: Job
        metadata:
          name: "{{ job_name }}"
        spec:
          backoffLimit: 0
          template:
            spec:
              restartPolicy: Never
              containers:
                - name: migrate
                  image: "busybox:1.36"
                  command: ["/bin/sh", "-c", "echo migrating; sleep 5"]
    output_name: created

k8s.apply runs once — it is not retried

Its default policy is max_attempts: 1. Server-side apply is itself idempotent, but a multi- document bundle that fails at document 5 would re-apply documents 1–4 on a retry. If your bundle is a single document, or purely declarative, raise max_attempts per call.

---

k8s.get

Fetches a single resource.

Input

Field	Type	Required	Default	Description
connection	object	yes	—	Cluster connection
api_version	str	no	"v1"	e.g. "apps/v1"
kind	str	yes	—	e.g. "Deployment"
name	str	yes	—	Resource name
namespace	str	no	connection's	Ignored for cluster-scoped kinds
not_found_ok	bool	no	false	Return found: false instead of failing

Output

Field	Type	Description
found	bool	Whether the resource exists
resource	object | null	The full resource
api_version, kind, name	str	Echoed back from the request
namespace	str | null	Namespace requested; null when cluster-scoped

Use not_found_ok: true when the question is "does this exist?" rather than "fetch this".

---

k8s.list

Lists resources of a kind.

Input

Field	Type	Required	Default	Description
connection	object	yes	—	Cluster connection
api_version	str	no	"v1"	e.g. "apps/v1"
kind	str	yes	—	Kind to list
namespace	str	no	connection's	—
all_namespaces	bool	no	false	List cluster-wide instead
label_selector	str	no	null	e.g. "app=api,tier!=canary"
field_selector	str	no	null	e.g. "status.phase=Running"
limit	int	no	500	Page size, capped at 2000
continue_token	str	no	null	From a previous call's output

Output

Field	Type	Description
items	list	Matched resources
item_count	int	Items in this page
continue_token	str | null	Pass back to fetch the next page; null when complete
resource_version	str | null	resourceVersion of the list

Every item lands in workflow history

A pod list in a busy namespace is megabytes, copied into history and replayed on every workflow task. Narrow with selectors and keep limit small; project what you need with output_data.

---

k8s.delete

Deletes a resource by name, or a set of them by label selector.

Input

Field	Type	Required	Default	Description
connection	object	yes	—	Cluster connection
api_version	str	no	"v1"	e.g. "apps/v1"
kind	str	yes	—	Kind to delete
name	str	one-of	—	Exclusive with label_selector
label_selector	str	one-of	—	Requires allow_bulk_delete
namespace	str	no	connection's	—
allow_bulk_delete	bool	no	false	Required for a selector delete
allow_cluster_scoped	bool	no	false	Required for a cluster-scoped kind
allow_system_namespaces	bool	no	false	Permit kube-system and friends
grace_period_seconds	int	no	null	0 forces immediate deletion
propagation_policy	str	no	"Background"	Foreground, Background or Orphan
not_found_ok	bool	no	true	Treat an absent resource as success

Output

Field	Type	Description
deleted	bool	Whether anything was deleted
deleted_count	int	Number of resources deleted
names	list[str]	Names of the deleted resources
not_found	bool	True when the target was already absent

not_found_ok defaults to true here (unlike k8s.get) — deleting something already gone is what makes a cleanup step safe to re-run. Default policy is max_attempts: 1.

---

k8s.scale

Sets the replica count on a Deployment, StatefulSet or ReplicaSet through the scale subresource, and reports what the count was before.

Input

Field	Type	Required	Default	Description
connection	object	yes	—	Cluster connection
kind	str	no	"Deployment"	Deployment, StatefulSet or ReplicaSet
name	str	yes	—	Workload name
namespace	str	no	connection's	—
replicas	int	yes	—	Desired count, >= 0

Output

Field	Type	Description
kind, name, namespace	str	Echoed back
previous_replicas	int | null	Count before the change — lets you report the delta or detect a no-op
replicas	int	Count now requested

Scaling only requests the change. Follow it with k8s.wait to block until the pods are actually ready.

---

k8s.logs

Reads a bounded tail of a pod's log.

Input

Field	Type	Required	Default	Description
connection	object	yes	—	Cluster connection
pod_name	str	one-of	—	Exclusive with label_selector
label_selector	str	one-of	—	Read every matching pod
max_pods	int	no	5	Cap when using a selector
namespace	str	no	connection's	—
container	str	no	null	Defaults to the pod's default container
tail_lines	int	no	200	Lines from the end of the log
since_seconds	int	no	null	Only lines newer than this
previous	bool	no	false	Read the previous terminated container's log
timestamps	bool	no	false	Prefix each line with an RFC3339 timestamp
max_bytes	int	no	1048576	Per-pod cap; the tail is kept when truncating

Output

Field	Type	Description
pods	list	One entry per pod read
pods[].pod_name	str	Pod the log came from
pods[].container	str | null	Container, when one was named
pods[].log	str	The log text
pods[].truncated	bool	Whether it was cut to fit max_bytes
pods[].byte_count	int	Size of the returned text
pod_count	int	Number of pods read

label_selector is how you reach a Job's pod, whose name the cluster generates:

- activity:
    type: k8s.logs
    name: collect-logs
    input_data:
      connection: "{{ k8s }}"
      label_selector: "job-name={{ job_name }}"
      tail_lines: 200
    output_name: job_logs

There is no follow/streaming mode

Pod logs are a high-throughput data feed, and Moco's event relay is built for low-rate control streams. Wait for the resource with k8s.wait, then read the tail. For progress during a long run, poll k8s.logs.

---

k8s.wait

Polls a resource until a condition holds. This is the activity that turns "I asked for a rollout" into "the rollout is live".

Exactly one of three condition forms is required:

1. condition — a status.conditions[].type such as Available, Ready, Complete or any custom-resource condition, matched against condition_status (default "True").
2. jsonpath + value — a dotted path such as status.readyReplicas, compared as a string. List indices are supported: spec.containers.0.image.
3. deleted: true — satisfied when the resource is gone.

Input

Field	Type	Required	Default	Description
connection	object	yes	—	Cluster connection
api_version	str	no	"v1"	e.g. "apps/v1"
kind	str	yes	—	Kind to wait on
name	str	yes	—	Resource name
namespace	str	no	connection's	—
condition	str	one-of	—	Condition type to wait for
condition_status	str	no	"True"	Status that condition must reach
jsonpath	str	one-of	—	Dotted path, compared against value
value	str	with jsonpath	—	Expected value
deleted	bool	one-of	false	Wait for the resource to disappear
fail_on_conditions	list[str]	no	null	Condition types that mean it can never succeed
poll_interval_sec	float	no	2.0	Initial gap; backs off to a 15s ceiling
raise_on_timeout	bool	no	true	Set false to get met: false back and branch

Output

Field	Type	Description
met	bool	Whether the condition was satisfied
elapsed_sec	float	Wall-clock seconds spent waiting
poll_count	int	Number of polls performed
observed	str | null	Last observed condition status or path value
resource	object | null	The resource as last seen

fail_on_conditions is what stops a failed Job from consuming the whole budget:

- activity:
    type: k8s.wait
    name: await-job
    input_data:
      connection: "{{ k8s }}"
      api_version: batch/v1
      kind: Job
      name: "{{ job_name }}"
      condition: Complete
      fail_on_conditions: ["Failed"]
      raise_on_timeout: false
    retry_policy:
      timeout_sec: 900
    output_name: awaited

This activity blocks, and heartbeats while it does

Its default budget is 600s with Temporal heartbeats configured, so a worker crash mid-wait is rescheduled rather than silently lost. If the workflow has other work to do meanwhile, put it in a parallel branch or set async_mode: true on the call.

A condition that is never met raises a non-retryable error, so a genuine failure is not multiplied by max_attempts.

---

k8s.exec

Runs a command inside a pod's container and captures its output and exit status.

Input

Field	Type	Required	Default	Description
connection	object	yes	—	Cluster connection
pod_name	str	yes	—	Pod to run in
namespace	str	no	connection's	—
container	str	no	null	Defaults to the pod's default container
command	list[str]	yes	—	Command and arguments. Not a string
stdin	str	no	null	Text piped to the command's standard input
timeout_sec	int	no	60	Budget for the remote command
max_output_bytes	int	no	1048576	Cap on each of stdout and stderr

Output

Field	Type	Description
stdout	str	Captured standard output
stderr	str	Captured standard error
exit_code	int	Exit status; 0 means success
truncated	bool	Whether either stream was cut

- activity:
    type: k8s.exec
    name: check-migration-state
    input_data:
      connection: "{{ k8s }}"
      pod_name: "{{ pod }}"
      command: ["/bin/sh", "-c", "psql -tAc 'select count(*) from schema_migrations'"]
    output_name: migration_check

k8s.exec is arbitrary remote code execution

It is the cluster equivalent of shell.run, and should be governed by the exec action on internal.k8s_activities. Unlike shell.run it defaults to max_attempts: 1, because an arbitrary command is not idempotent.

command is a list and is never passed through a shell. To use one, say so explicitly: ["/bin/sh", "-c", "..."].

---

Why there is no k8s.job_run

Running a batch Job is the most common reason a workflow talks to a cluster, so a single create-wait-logs-cleanup activity is a natural thing to want. It is not offered, because the four phases have four different retry postures and one activity manifest carries only one retry policy. Fused together, an API server blip while fetching logs would fail the activity even though the Job had already succeeded — and the workflow could not tell the difference.

Composed in a wfspec, each step retries on its own terms, you can see which phase failed, and "clean up on success but keep the evidence on failure" is a single condition:. The full worked example is moco-examples/k8s-demo/src/k8s-job-run.yaml:

- activity: {type: k8s.apply, name: create-job,   ...}
- activity: {type: k8s.wait,  name: await-job,    ...}
- activity: {type: k8s.logs,  name: collect-logs, ...}
- activity:
    type: k8s.delete
    name: cleanup-job
    condition: "{{ cleanup and awaited['met'] }}"
    input_data: {...}