Buffering and data loss prevention on k8s

[asafdl]

Hi, I'm looking to configure buffering for my sink, with the intent of having 0 data loss. I saw in the docs that you suggest not using aws gp2 (what we are using) for disk buffer, are there any other suggestions for what I should use?

Also the disk buffer doesn't specify what volume it is loaded on, is there a way I can use a persistent volume and not the node volume for this?

Also is there any info on how hard is the performance hit for using disk buffer? I know it will be dependent on a bunch of things like type of disk + cpu etc. But is this on a scale of milliseconds, seconds or minutes difference?

Thanks.

[binarylogic]

Hey @asafdl great question! We're working on a set of comprehensive guides that'll cover questions like this and help with going to production with large deployments.

That said, could you provide a little more info about your data sources and sinks? In general, disk buffers should be a last resort for situations that do not offer alternative durability strategies. I would not use disk buffers if you absolutely cannot lose data. Better strategies include:

1. Using Vector's new end-to-end acknowledgement feature. This ensures that Vector does not advance reading data (sources) without confirmation that data was sent downstream (sinks). This pushes the responsibility of durability upstream, which is generally much better for performance. This feature is in beta, but our confidence is pretty high in its reliability. You can see the docs here.
2. Leverage systems designed for the level of durability you require. The obvious suggestion is Kafka/Kinesis, but we understand that is a very large ask of someone.
3. Finally, we are working through enterprise-level data loss strategies, like using other systems for external buffering (e.g. s3) and features like dead letter queues (routing failed events to another storage). We are expected complete these in Q3/Q4 of this year. If you're interested in being an early tester, please reach out to me (ben.johnson@datadoghq.com) and we can chat.

I hope that helps.

[joshlreese]

@binarylogic - Thank you for the detailed explanation here. One question regarding disk durability - would you consider a volume backend which ensures data durability (assuming a flush has occurred) and allows for node-mobility of an instance of Vector to be a downgrade in durability?

[JustinMason]

I am seeing data loss in a very basic example. I assume I am missing a setting.

We have 2 processes.
"Router" sources from Kafka, sinks to Vector
"Sink" sources from Vector, sinks back to Kafka

"Router"

      sources:
        kafka-in:
          type: kafka
          acknowledgements: true
          bootstrap_servers: logstreaming-kafka-bootstrap.kafka:9092
          group_id: router-consumer3
          topics:
            - vector.test.in
      transforms:
        timestamper:
          type: remap
          inputs:
            - kafka-in
          source: |-
            . = parse_json!(.message)
            .t1 = to_unix_timestamp(now())
      sinks:
        vector-sink:
          type: vector
          inputs:
            - timestamper
          address: sink:9000
          version: "2"
          buffer:
            type: disk
            max_size: 10485760
            when_full: block

"Sink"

  sources: 
        vector_router_source:
          type: vector
          acknowledgements: true
          version: "2"
          address: 0.0.0.0:9000
      transforms:
        timestamper:
          type: remap
          inputs:
            - vector_router_source
          source: |-
            .t2 = to_unix_timestamp(now())
      sinks:
        sink:
          type: kafka
          inputs:
            - timestamper
          bootstrap_servers: logstreaming-kafka-bootstrap.kafka:9092
          encoding:
            codec: json
          healthcheck:
            enabled: true
          topic: vector.test.out
          buffer:
            type: disk
            max_size: 10485760
            when_full: block

This is running in GKE. When we intentionally delete the "Sink" during a run we expect it to come back and resume from where it left off. We are seeing gaps in whats "Sinked" back to Kafka. When we don't delete the Sink, happy path, all the data sinks back to Kafka.
I have the acknowledgements, the Buffer Set to disk, and we are deploying to GKE with a persistent volume claim for both the "Router" and the "Sink", see router yaml section below:

      name: vector
        ports:
        - containerPort: 9000
        resources:
          limits:
            cpu: 2
            memory: 6000Mi
          requests:
            cpu: 2
            memory: 6000Mi
        volumeMounts:
        - mountPath: /etc/vector/
          name: config
        - mountPath: /var/lib/vector/
          name: vector-router-data
      volumes:
      - configMap:
          name: router-config
        name: config
      - name: vector-router-data
        persistentVolumeClaim:
          claimName: vector-router-data
  volumeClaimTemplates:
  - metadata:
      name: vector-router-data
    spec:
      accessModes:
      - ReadWriteOnce
      resources:
        requests:
          storage: 10Gi

Am I missing something obvious? I would assume that the "Router" process would not get acknowledgements from the "Sink" process and it would replay/buffer the data once the "Sink" resumed.

I'm seeing errors from the Router when I delete the Sink.
:30.197 ERROR sink{component_kind="sink" component_name=vector-sink component_type=vector}:request{request_id=13548}: vector::sinks::util::retries: Non-retriable error; dropp │ │ Aug 10 02:47:30.198 ERROR sink{component_kind="sink" component_name=vector-sink component_type=vector}:request{request_id=13548}: vector::sinks::util::sink: Request failed. error=Request │

Any ideas?
Thanks

[spencergilbert]

Tagging @bruceg for to the end-to-end acknowledgements

[tdorianh]

More info regarding the above, with vector v2 sink -> vector v2 source, acknowledgements=true--I've replicated the issue to capture the complete error messages (output log is from the vector v2 sink section):

Aug 10 16:25:47.187 ERROR sink{component_kind="sink" component_name=vector-sink component_type=vector}:request{request_id=123}: vector::sinks::util::retries: Non-retriable error; dropping the request. error=Request failed: status: Unknown, message: "transport error: operation was canceled: connection closed", details: [], metadata: MetadataMap { headers: {} }
Aug 10 16:25:47.193 ERROR sink{component_kind="sink" component_name=vector-sink component_type=vector}:request{request_id=123}: vector::sinks::util::sink: Request failed. error=Request failed: status: Unknown, message: "transport error: operation was canceled: connection closed", details: [], metadata: MetadataMap { headers: {} }
Aug 10 16:25:47.199 ERROR sink{component_kind="sink" component_name=vector-sink component_type=vector}:request{request_id=124}: vector::sinks::util::retries: Non-retriable error; dropping the request. error=Request failed: status: Unknown, message: "transport error: error trying to connect: tcp connect error: Connection refused (os error 111)", details: [], metadata: MetadataMap { headers: {} }
Aug 10 16:25:47.200 ERROR sink{component_kind="sink" component_name=vector-sink component_type=vector}:request{request_id=124}: vector::sinks::util::sink: Request failed. error=Request failed: status: Unknown, message: "transport error: error trying to connect: tcp connect error: Connection refused (os error 111)", details: [], metadata: MetadataMap { headers: {} }
Aug 10 16:25:55.220  WARN sink{component_kind="sink" component_name=vector-sink component_type=vector}:request{request_id=161}: vector::sinks::util::retries: Retrying after error. error=Request failed: status: Internal, message: "h2 protocol error: protocol error: not a result of an error", details: [], metadata: MetadataMap { headers: {} }
Aug 10 16:25:55.248 ERROR sink{component_kind="sink" component_name=vector-sink component_type=vector}:request{request_id=162}: vector::sinks::util::retries: Non-retriable error; dropping the request. error=Request failed: status: Unknown, message: "transport error: error trying to connect: tcp connect error: Connection refused (os error 111)", details: [], metadata: MetadataMap { headers: {} }

Looks like the data loss may be caused by the vector v2 sink not retrying on connection closed or refused.

[bruceg]

Thanks for diagnosing this, @tdorianh. I can confirm that the vector v2 sink is not retrying on refused connections. I have opened up a bug issue for this: #8648