Pipeline/vertices pods are not starting after a Kubernetes cluster restart.

[ladocz]

The question is: how can we debug this situation? Please let me know if anyone has any ideas on how to analyze such a bug.
After a restart, all pipeline pods remain in the "Init" state and do not start.
On the Pipeline Daemon, I receive the following message, but it doesn't provide much help:

stream logs failed container "numa" in pod "numaflow-pipeline-daemon-77758cf879-c9js4" is waiting to start: PodInitializing for inqu-analytics/numaflow-pipeline-daemon-77758cf879-c9js4 (numa)
init {"level":"info","ts":"2025-01-17T17:05:33.106947505Z","logger":"numaflow.isbsvc-validate","caller":"commands/isbsvc_validate.go:81",
"msg":"Buffers, buckets and side inputs store might have not been created yet, will retry if the limit is not reached","pipeline":"numaflow-pipeline",
"error":"failed to query OT KV \"inqu-analytics-numaflow-pipeline-nats-endpoint-merge_OT\", nats: bucket not found"}

If I delete pipeline in Numaflow Web UI. And create the same pipepeline again. It works fine.

[vigith]

• which version of numaflow are you using?
• Do you use ISB with PVC backed volumes?
• could you please share your pipeline and ISB spec?

[ladocz]

Currently I am on numalfow v 1.4.0.
I am using ISB with persistens -PVC with my default storage class.

ISB Spec:

apiVersion: numaflow.numaproj.io/v1alpha1
kind: InterStepBufferService
metadata:
  name: default
  namespace: inqu-analytics
spec:
  jetstream:
    replicas: 1
    settings: |
      max_payload: 8388608
      max_memory_store: 8073741824
    version: 2.10.17
    persistence:
      volumeSize: 1Gi
      #storageClassName: local-path # Optional, will use K8s cluster default storage class if not specified  
    bufferConfig: |
      # The properties of the buffers (streams) to be created in this JetStream service
      stream:
        # 0: Limits, 1: Interest, 2: WorkQueue
        retention: 2
        maxAge: 1h
        # 0: File, 1: Memory
        storage: 0
        maxMsgs: 1000000
      # The consumer properties for the created streams
      consumer:
        ackWait: 1h
        maxAckPending: 1000000
        replicas: 1
        storage: 1
      otBucket:
        maxValueSize: 0
        history: 1
        ttl: 24h
        maxBytes: 0
        storage: 1         ### <- this switch to RAM only mode
        replicas: 1        ### <- this does not replicate data 
      procBucket:
        maxValueSize: 0
        history: 1
        ttl: 24h
        maxBytes: 0
        storage: 1         ### <- this switch to RAM only mode
        replicas: 1        ### <- this does not replicate data 

and my pipeline

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: numaflow-pipeline
  namespace: inqu-analytics
spec:
  limits:
    bufferMaxLength: 1000000
  vertices:
    - name: source1
      scale:
        disabled: true
      source:
        udsource:
          container:
            image: drswvs-proget1.rosen-nxt.com/docker-inqu/inqu-numa-source:0.0.58-dev # {"$imagepolicy": "flux-system:inqu-numa-source"}
            imagePullPolicy: Always
            env:
              - name: CONFIGURATION_PREFECT_BLOCK
                value: "pipeline-source1-configuration"
              - name: PREFECT_LOGGING_LEVEL
                value: "WARNING"
    - name: source2
      scale:
        disabled: true
      source:
        udsource:
          container:
            image: drswvs-proget1.rosen-nxt.com/docker-inqu/inqu-numa-source:0.0.58-dev # {"$imagepolicy": "flux-system:inqu-numa-source"}
            imagePullPolicy: Always
            env:
              - name: CONFIGURATION_PREFECT_BLOCK
                value: "pipeline-source2-configuration"                     
              - name: PREFECT_LOGGING_LEVEL
                value: "WARNING"
                # limits:
                #   readBatchSize: 
    - name: nats-endpoint
      source:
        nats:
          url: nats:4222
          subject: analytics
          queue: analytics
          auth:
            token:
              name: numaflow-token
              key: token
              # transformer:
              #   container:
              #     image: drswvs-proget1.rosen-nxt.com/docker-inqu/inqu-numa-transform:0.0.58-feature_numaflow
              #     imagePullPolicy: Always
      # limits:
      #   readBatchSize: 1
    - name: http-endpoint
      source:
        http:
          service: True
          auth:
            token:
              name: numaflow-token
              key: token
    - name: merge
      partitions: 1
      scale:
        disabled: true
      udf:
        container:
          image: drswvs-proget1.rosen-nxt.com/docker-inqu/inqu-numa-merge:0.0.58-dev # {"$imagepolicy": "flux-system:inqu-numa-merge"}
          imagePullPolicy: Always
          env:
            - name: MAX_BATCH_SIZE
              value: "1000"
        groupBy:
          window:
            fixed:
              length: 5s
              streaming: true # set streaming to true to enable reduce streamer              
          storage:
            persistentVolumeClaim:
              volumeSize: 1Gi
              accessMode: ReadWriteOnce
    - name: map
      scale:
        disabled: true
      udf:
        container:
          image: drswvs-proget1.rosen-nxt.com/docker-inqu/inqu-numa-map:0.0.58-dev # {"$imagepolicy": "flux-system:inqu-numa-map"}
          imagePullPolicy: Always
          env:
            - name: INQU_LOGGING_LEVEL
              value: "ERROR"
    - name: sink
      scale:
        disabled: true
      sink:
        udsink:
          container:
            image: drswvs-proget1.rosen-nxt.com/docker-inqu/inqu-numa-sink:0.0.58-dev # {"$imagepolicy": "flux-system:inqu-numa-sink"}
            imagePullPolicy: Always
            env:
              - name: PREFECT_LOGGING_LEVEL
                value: "WARNING"
        fallback:
          log: {}
    - name: out-log
      scale:
        disabled: true
      sink:
        log: {}
  edges:
    - from: nats-endpoint
      to: merge
      onFull: discardLatest
    - from: source1
      to: map
      onFull: discardLatest
    - from: http-endpoint
      to: map
      onFull: discardLatest
    - from: source2
      to: map
      onFull: discardLatest
    - from: merge
      to: map
      onFull: discardLatest
    - from: map
      to: sink
      conditions:
        tags:
          values:
            - dwh
      onFull: discardLatest
    - from: map
      to: out-log
      conditions:
        tags:
          values:
            - unknown
      onFull: discardLatest
  watermark:
    idleSource:
      threshold: 15s # The pipeline will be considered idle if the source has not emitted any data for given threshold value.
      incrementBy: 3s # If source is found to be idle then increment the watermark by given incrementBy value.
      stepInterval: 2s # If source is idling then publish the watermark only when step interval has passed.

[whynowy]

Can you check the status of the pvc and pv after cluster restart? Is the pv still existing? old? new? And what the storage class used by the pv?

[ladocz]

Status of my pv and pvc after cluster restart is Bound, Age of both after restart are more days. In my eyes it looks good to me. And the storage class is local-path from rancher.io
pv describe:

Name:              pvc-d123d2fa-b9b6-48d4-be1f-254ef31b537d
Labels:            <none>
Annotations:       local.path.provisioner/selected-node: drswpc-bz70c14
                   pv.kubernetes.io/provisioned-by: rancher.io/local-path
Finalizers:        [kubernetes.io/pv-protection]
StorageClass:      local-path
Status:            Bound
Claim:             inqu-analytics/isbsvc-default-js-vol-isbsvc-default-js-0
Reclaim Policy:    Delete
Access Modes:      RWO
VolumeMode:        Filesystem
Capacity:          1Gi
Node Affinity:     
  Required Terms:  
    Term 0:        kubernetes.io/hostname in [drswpc-bz70c14]
Message:           
Source:
    Type:  LocalVolume (a persistent volume backed by local storage on a node)
    Path:  /var/lib/rancher/k3s/storage/pvc-d123d2fa-b9b6-48d4-be1f-254ef31b537d_inqu-analytics_isbsvc-default-js-vol-isbsvc-default-js-0
Events:    <none>

pvc-describe

Name:          isbsvc-default-js-vol-isbsvc-default-js-0
Namespace:     inqu-analytics
StorageClass:  local-path
Status:        Bound
Volume:        pvc-d123d2fa-b9b6-48d4-be1f-254ef31b537d
Labels:        app.kubernetes.io/component=isbsvc
               app.kubernetes.io/managed-by=isbsvc-controller
               app.kubernetes.io/part-of=numaflow
               numaflow.numaproj.io/isbsvc-name=default
               numaflow.numaproj.io/isbsvc-type=jetstream
Annotations:   pv.kubernetes.io/bind-completed: yes
               pv.kubernetes.io/bound-by-controller: yes
               volume.beta.kubernetes.io/storage-provisioner: rancher.io/local-path
               volume.kubernetes.io/selected-node: drswpc-bz70c14
               volume.kubernetes.io/storage-provisioner: rancher.io/local-path
Finalizers:    [kubernetes.io/pvc-protection]
Capacity:      1Gi
Access Modes:  RWO
VolumeMode:    Filesystem
Used By:       isbsvc-default-js-0
Events:        <none>

default storage class:

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  annotations:
    defaultVolumeType: local
    objectset.rio.cattle.io/id: ""
    objectset.rio.cattle.io/owner-gvk: k3s.cattle.io/v1, Kind=Addon
    objectset.rio.cattle.io/owner-name: local-storage
    objectset.rio.cattle.io/owner-namespace: kube-system
    storageclass.kubernetes.io/is-default-class: "true"
  creationTimestamp: "2025-01-14T14:05:57Z"
  labels:
    objectset.rio.cattle.io/hash: 183f35c65ffbc3064603f43f1580d8c68a2dabd4
  name: local-path
  resourceVersion: "281"
  uid: 1a592580-829a-42e7-b6fb-6b36a648f492
provisioner: rancher.io/local-path
reclaimPolicy: Delete
volumeBindingMode: WaitForFirstConsumer

[whynowy]

This seems to be a jetstream problem: when it's not running in cluster mode (single node), if the stream or k/v bucket uses RAM storage mode, after restarting, the stream or k/v bucket will be lost.

What you can do is, either use cluster mode (e.g. 3 replicas), or use file storage with 1 replica.

[ladocz]

Thanks for your suggestions. Both suggestions have worked and
ISB Settings for both of them I have put bellow for other users to follow.

What I do not understand is why the memory storage doesn´t work with non cluster ISB Service. Is it a Bug in Nats JetStream or is there any reason behind.

1. I have used file storage with 1 replica - no cluster. See the jetstream settings (only relevant changes):

apiVersion: numaflow.numaproj.io/v1alpha1
kind: InterStepBufferService
metadata:
  name: default
  namespace: inqu-analytics
spec:
  jetstream:
    replicas: 1    
    persistence:
      volumeSize: 1Gi
      #storageClassName: local-path # Optional, will use K8s cluster default storage class if not specified  
    bufferConfig: |
      # The properties of the buffers (streams) to be created in this JetStream service
      stream:
        # 0: File, 1: Memory
        storage: 0
      consumer:
        replicas: 1
        storage: 1      ### <- this switch to file storage
      otBucket:
        storage: 1         ### <- this switch to file storage
        replicas: 1        ### <- this does not replicate data 
      procBucket:
        storage: 1         ### <- this switch to file storage
        replicas: 1        ### <- this does not replicate data 

2. Memory storage and with 3 replicas cluster. See the jetstream settings (only relevant changes):

apiVersion: numaflow.numaproj.io/v1alpha1
kind: InterStepBufferService
metadata:
  name: default
  namespace: inqu-analytics
spec:
  jetstream:
    replicas: 3      ### <- 3 replicas cluster
    persistence:
      volumeSize: 1Gi
      #storageClassName: local-path # Optional, will use K8s cluster default storage class if not specified  
    bufferConfig: |
      # The properties of the buffers (streams) to be created in this JetStream service
      stream:
        # 0: File, 1: Memory
        storage: 0
      consumer:
        replicas: 0
        storage: 1      ### <- this switch to file storage
      otBucket:
        storage: 0         ### <- this switch to file storage
        replicas: 1        ### <- this does not replicate data 
      procBucket:
        storage: 0         ### <- this switch to file storage
        replicas: 1        ### <- this does not replicate data 

[syayi]

Hi @ladocz ! Apologies for jumping into the discussion. Thank you for exploring Numaflow! Would love to hear how you discovered the project? Are there any specific use cases you're considering for it?

[ladocz]

We are using Numaflow as a data streaming platform to collect data from external sources into our DWH.
Some data sources are databases, and we also have plans with HTTP or NATS sources. As a sink, we again use a database. We also have some UDF transformations in between.
We are still in the incubation/development phase with Numaflow, but I am convinced this was a good choice for our data streaming pipeline.
I like the Numaflow features: Kubernetes-native with easy deployment and scaling, NATS for ISB, and the ability to customize Vertices nodes with the Python SDK. I am also curious about the Numalogic ML project, though I haven't tried it yet.
In short, the Numaflow team has done a great job!

[syayi]

Hi @ladocz! Thank you for sharing your use case and feedback—it’s amazing to hear how you’re leveraging Numaflow! 🙌 We’d love to have you join our Slack. We’re also hosting community meetings soon, and it’d be great to have all the community join us