Enroll workers on Kubernetes

Workers run the leader's image, named by the same digest, and connect to a leader deployed as in Run the leader on Kubernetes. Workers run either as a StatefulSet, each with its own persistent identity, or as a Deployment that an autoscaler grows and shrinks, each pod enrolling itself when it starts.

Deploy a worker

First create and deploy the worker group in Smeltr, then mint a permanent enrollment code that brings in enough workers for the pods you are starting. Save this as smeltr-worker.yaml and replace the image digest, leader URL and enrollment code:

apiVersion: v1
kind: ConfigMap
metadata:
  name: smeltr-worker-default
data:
  smeltr.toml: |
    [worker]
    enabled = true
    group = "default"

    [worker.leader]
    url = "https://smeltr.example.com"

    [worker.storage]
    state_dir = "/var/lib/smeltr-worker"

    [worker.runtime]
    thread_limit = 2
---
apiVersion: v1
kind: Secret
metadata:
  name: smeltr-worker-default-enrollment-code
type: Opaque
stringData:
  enrollment-code: PASTE_THE_COMPLETE_ENROLLMENT_CODE_HERE
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: smeltr-worker-default
spec:
  serviceName: smeltr-worker-default
  replicas: 1
  selector:
    matchLabels:
      app: smeltr-worker-default
  template:
    metadata:
      labels:
        app: smeltr-worker-default
    spec:
      terminationGracePeriodSeconds: 430
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        fsGroup: 1000
      containers:
        - name: smeltr
          image: ghcr.io/smeltr-sh/smeltr@sha256:REPLACE_WITH_DIGEST
          imagePullPolicy: IfNotPresent
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: [ALL]
            seccompProfile:
              type: RuntimeDefault
          env:
            - name: SMELTR_CONFIG_FILE
              value: /config/smeltr.toml
            - name: SMELTR_WORKER_ENROLLMENT_CODE
              valueFrom:
                secretKeyRef:
                  name: smeltr-worker-default-enrollment-code
                  key: enrollment-code
          resources:
            requests:
              cpu: "2"
              memory: 2Gi
            limits:
              memory: 2Gi
          volumeMounts:
            - name: config
              mountPath: /config
              readOnly: true
            - name: worker-state
              mountPath: /var/lib/smeltr-worker
      volumes:
        - name: config
          configMap:
            name: smeltr-worker-default
  volumeClaimTemplates:
    - metadata:
        name: worker-state
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 2Gi

The StatefulSet gives each numbered worker pod its own persistent volume. Never share or clone those claims: each contains one worker identity and credential. Workers require outbound access to the public leader URL and to their configured sources and destinations. They require no Service or inbound port unless their group runs a receiver; see Expose a receiver.

thread_limit matches the CPU request, since a pod with no CPU limit otherwise runs a transform thread for every core on its node. Size the memory limit for what the group runs: see Size a worker.

terminationGracePeriodSeconds gives a stopping worker time to stop cleanly. It gives its pipeline up to 90 seconds to drain before killing it, then tells the leader it is stopping, then exits. Most stops take seconds, and the pod ends when the worker exits. Every step is bounded. The slowest stop takes about seven minutes: a pipeline the worker must kill, a stop that lands mid-deploy, and a leader slow to answer. 430 seconds covers the slowest stop. Status shows a worker that stops this way as Stopped, and it raises no warning. A shorter grace period can end a worker mid-drain: what it read since its last commit is read again by whichever worker next owns it, and the leader reports the worker silent.

After the worker appears Online, remove the SMELTR_WORKER_ENROLLMENT_CODE environment entry from the StatefulSet, apply the change, and confirm the restarted pod reconnects using its persisted credential. Then delete the enrollment code Secret. The worker needs it only to enroll.

Autoscale workers

A group that grows and shrinks with load runs as a Deployment. Every pod enrolls itself with one ephemeral enrollment code, and the leader forgets each worker once it is gone. Create and deploy the group, then in Admin → Workers → Enrollment codes choose + Add enrollment code, set Workers it brings in to Ephemeral and select Mint. An ephemeral enrollment code has no expiry and no limit on the workers it brings in: it works until you withdraw it. Save this as smeltr-worker-ingest.yaml and replace the image digest, leader URL and enrollment code:

apiVersion: v1
kind: ConfigMap
metadata:
  name: smeltr-worker-ingest
data:
  smeltr.toml: |
    [worker]
    enabled = true
    group = "ingest"

    [worker.leader]
    url = "https://smeltr.example.com"

    [worker.storage]
    state_dir = "/var/lib/smeltr-worker"

    [worker.runtime]
    thread_limit = 2
---
apiVersion: v1
kind: Secret
metadata:
  name: smeltr-worker-ingest-enrollment-code
type: Opaque
stringData:
  enrollment-code: PASTE_THE_COMPLETE_EPHEMERAL_ENROLLMENT_CODE_HERE
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: smeltr-worker-ingest
spec:
  replicas: 2
  selector:
    matchLabels:
      app: smeltr-worker-ingest
  template:
    metadata:
      labels:
        app: smeltr-worker-ingest
    spec:
      terminationGracePeriodSeconds: 430
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
        fsGroup: 1000
      containers:
        - name: smeltr
          image: ghcr.io/smeltr-sh/smeltr@sha256:REPLACE_WITH_DIGEST
          imagePullPolicy: IfNotPresent
          securityContext:
            allowPrivilegeEscalation: false
            readOnlyRootFilesystem: true
            capabilities:
              drop: [ALL]
            seccompProfile:
              type: RuntimeDefault
          env:
            - name: SMELTR_CONFIG_FILE
              value: /config/smeltr.toml
            - name: SMELTR_WORKER_ENROLLMENT_CODE
              valueFrom:
                secretKeyRef:
                  name: smeltr-worker-ingest-enrollment-code
                  key: enrollment-code
          resources:
            requests:
              cpu: "2"
              memory: 2Gi
            limits:
              memory: 2Gi
          volumeMounts:
            - name: config
              mountPath: /config
              readOnly: true
            - name: worker-state
              mountPath: /var/lib/smeltr-worker
      volumes:
        - name: config
          configMap:
            name: smeltr-worker-ingest
        - name: worker-state
          emptyDir: {}
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: smeltr-worker-ingest
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: smeltr-worker-ingest
  minReplicas: 2
  maxReplicas: 12
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70

Keep the enrollment code Secret and its environment entry: every new pod enrolls with it. The state directory is an emptyDir, so a worker's identity lives and dies with its pod. A container that restarts in its pod keeps its identity; a pod that is replaced enrolls as a new worker, and clones the leader's configuration repository before it runs anything.

A worker that stops cleanly tells the leader, which forgets it at once. The leader forgets a worker that is killed without stopping cleanly once it has been silent for the time set under Admin → Workers → Ephemeral workers, ten minutes by default. A worker the leader has forgotten that comes back, such as one cut off from the leader for longer than that, enrolls again with the enrollment code as a new worker and restarts as it.

  • Keep maxReplicas at or below the partition count of the Kafka topics the group reads. A worker beyond that is assigned no partitions and sits idle.
  • Do not autoscale a group whose Kafka source reads a selected range, one with a start or stop position. A range read shares no work: every pod reads the whole range.
  • Set a CPU request for the HorizontalPodAutoscaler to scale on.
  • The enrollment code enrolls a worker, and an enrolled worker can receive every credential the leader stores. Restrict who can read the Secret. If it is exposed, withdraw it in the Enrollment codes table and mint another: running workers keep their credentials, and new pods need the new one.
  • Rotating the leader's certificate authority withdraws the enrollment code at cutover, and you cannot mint one while the rotation is open. Mint a new ephemeral enrollment code after cutover and update the Secret; until then new pods cannot enroll.

Expose a receiver

A receiver is a source that senders connect to. Every worker of its group listens on the receiver's port, so each worker pod needs that port open, and senders need one address that reaches all of them. Add the port to the worker container, beside its env. This example uses port 8088:

          ports:
            - name: receiver
              containerPort: 8088
          readinessProbe:
            httpGet:
              path: /health
              port: receiver
            periodSeconds: 5

Then add a Service that selects the worker pods:

apiVersion: v1
kind: Service
metadata:
  name: smeltr-worker-default-receiver
spec:
  type: LoadBalancer
  selector:
    app: smeltr-worker-default
  ports:
    - name: receiver
      port: 8088
      targetPort: receiver

Pick a Service type for your balancer. For an autoscaled group, select its Deployment's pods.

GET /health on the receiver port answers 200 while the receiver takes requests and 503 while it is full, blocked or stopping, so the readiness probe takes a busy worker out of the Service until it can take requests again. The probe is optional.

  • A receiver served under an identity speaks TLS, so its probe adds scheme: HTTPS.
  • Under an identity with authorities, every connection to the receiver must present a client certificate, /health included. An httpGet probe has none, so use a tcpSocket probe on the receiver port there. It confirms only that the worker listens.
  • The worker runs as a non-root user without capabilities and cannot listen on a port below 1024. For syslog's 514, set the receiver's port to a high one, such as 5514, and give the Service port: 514 with targetPort: receiver.
  • A syslog receiver over UDP needs protocol: UDP on its container port and on the Service's, and has no health path, so it takes no probe. Its allowFrom sees the address a datagram arrives from, so the Service keeps the sender's own: externalTrafficPolicy: Local on a LoadBalancer or NodePort Service.