Dataflow is a Kubernetes-native platform for executing large parallel data-processing pipelines.

Last update: Jan 4, 2023

Overview

Dataflow

Summary

Dataflow is a Kubernetes-native platform for executing large parallel data-processing pipelines.

Each pipeline is specified as a Kubernetes custom resource which consists of one or more steps which source and sink messages from data sources such Kafka, NATS Streaming, or HTTP services.

Each step runs zero or more pods, and can scale horizontally using HPA or based on queue length using built-in scaling rules. Steps can be scaled-to-zero, in which case they periodically briefly scale-to-one to measure queue length so they can scale a back up.

Learn more about features.

Use Cases

Real-time "click" analytics
Anomaly detection
Fraud detection
Operational (including IoT) analytics

Screenshot

Example

pip install git+https://github.com/argoproj-labs/argo-dataflow#subdirectory=dsls/python

from argo_dataflow import cron, pipeline

if __name__ == '__main__':
    (pipeline('hello')
     .namespace('argo-dataflow-system')
     .step(
        (cron('*/3 * * * * *')
         .cat()
         .log())
    )
     .run())

Documentation

Read in order:

Beginner:

Intermediate:

Advanced

Architecture Diagram

Comments

feat: Updating git step action to use credentials based authentication.

Both github & gitlab allow credentials or token based authentication to the repo. Since this project is using distroless images, I didn't want to add an ssh-agent dependency on the project, and thought it might be better to use token based authentication to load private repos.

opened by domderen 27
feat: Adding ability to specify imagePullSecrets for pipelines

This PR solves #183. I didn't add any tests or examples using it or docs. What do you think we should add for it? I tested it locally to confirm it's working.

opened by domderen 15
ArgoLabs Dataflow is on hold - use Numaflow

~We've put Dataflow on hold as we are planning on some significant changes and redesign. Stay tuned.~

Dataflow has been rewritten and is now available as Numaflow

opened by alexec 11
Pipeline is deleted but Steps remain

Using v0.0.115

I deleted the Pipeline via UI

There are no pipelines in the namespace but steps are there and so are the pods associated with those steps
bug good first issue no-issue-activity

opened by Udit107710 8

Multiple Sources to single Step

Scenario: A single step is consuming from two Kafka Topics and the step is failing with the attached trace in sidecar

Log trace:

time="2021-11-16T06:47:14Z" level=info msg="generating self-signed certificate"
time="2021-11-16T06:47:14Z" level=info msg="sidecar config" pipelineName=atlan-keycloak-role-bulk-rt replica=0 stepName=keycloak-role-sync updateInterval=15s
time="2021-11-16T06:47:14Z" level=info msg="connecting sink" sink="{\"name\":\"default\",\"kafka\":{\"name\":\"keycloak-role-sync-failed\",\"brokers\":[\"kafka-headless.kafka.svc.cluster.local:9092\"],\"topic\":\"keycloak-role-sync-failed\",\"batchSize\":\"100Ki\",\"compressionType\":\"lz4\",\"acks\":\"all\",\"enableIdempotence\":true}}"
time="2021-11-16T06:47:14Z" level=info msg="kafka config" config="{\"acks\":\"all\",\"batch.size\":102400,\"bootstrap.servers\":\"kafka-headless.kafka.svc.cluster.local:9092\",\"compression.type\":\"lz4\",\"enable.idempotence\":true,\"go.logs.channel.enable\":true,\"linger.ms\":0}" sink=default
time="2021-11-16T06:47:14Z" level=info msg="adding stop hook" sink=default
time="2021-11-16T06:47:14Z" level=info msg="FIFO out interface configured"
time="2021-11-16T06:47:14Z" level=info msg="HTTP out interface configured"
time="2021-11-16T06:47:14Z" level=info msg="HTTP in interface configured"
time="2021-11-16T06:47:14Z" level=info msg="waiting for HTTP in interface to be ready"
time="2021-11-16T06:47:14Z" level=info msg="consuming Kafka logs" sink=default
time="2021-11-16T06:47:14Z" level=info msg="starting producer event consuming loop" sink=default
time="2021-11-16T06:47:14Z" level=info msg="starting HTTP server"
time="2021-11-16T06:47:14Z" level=info msg="starting HTTPS server"
time="2021-11-16T06:47:14Z" level=info msg="HTTP in interface ready"
time="2021-11-16T06:47:14Z" level=info msg="connecting source" source="{\"name\":\"default\",\"kafka\":{\"name\":\"default\",\"brokers\":[\"kafka-headless.kafka.svc.cluster.local:9092\"],\"topic\":\"keycloak_intermediate_store\",\"startOffset\":\"Last\",\"fetchMin\":\"100Ki\",\"fetchWaitMax\":\"500ms\"},\"retry\":{\"duration\":\"100ms\",\"factorPercentage\":200,\"steps\":20,\"cap\":\"0s\",\"jitterPercentage\":10}}" urn="urn:dataflow:kafka:kafka-headless.kafka.svc.cluster.local:9092:keycloak_intermediate_store"
time="2021-11-16T06:47:14Z" level=info msg="Kafka config" config="{\"auto.offset.reset\":\"latest\",\"bootstrap.servers\":\"kafka-headless.kafka.svc.cluster.local:9092\",\"enable.auto.commit\":false,\"enable.auto.offset.store\":false,\"fetch.min.bytes\":102400,\"fetch.wait.max.ms\":500,\"go.logs.channel.enable\":true,\"group.id\":\"dataflow-bet-def-atl-key-def-cee67862d4d071a515ccddb184269d3103329b4471fee0805b5771ddbf2349f5\",\"group.instance.id\":\"dataflow-bet-def-atl-key-def-cee67862d4d071a515ccddb184269d3103329b4471fee0805b5771ddbf2349f5/0\",\"heartbeat.interval.ms\":3000,\"socket.keepalive.enable\":true,\"statistics.interval.ms\":5000}" source=default
time="2021-11-16T06:47:14Z" level=info msg="starting pending loop" source=default updateInterval=15s
time="2021-11-16T06:47:14Z" level=info msg="connecting source" source="{\"name\":\"default\",\"kafka\":{\"name\":\"keycloak-role-sync-retry\",\"brokers\":[\"kafka-headless.kafka.svc.cluster.local:9092\"],\"topic\":\"keycloak-role-sync-retry\",\"startOffset\":\"Last\",\"fetchMin\":\"100Ki\",\"fetchWaitMax\":\"500ms\"},\"retry\":{\"duration\":\"100ms\",\"factorPercentage\":200,\"steps\":20,\"cap\":\"0s\",\"jitterPercentage\":10}}" urn="urn:dataflow:kafka:kafka-headless.kafka.svc.cluster.local:9092:keycloak-role-sync-retry"
time="2021-11-16T06:47:14Z" level=info msg=pre-stop source=defer
time="2021-11-16T06:47:14Z" level=info msg="running hooks" len=2
time="2021-11-16T06:47:14Z" level=info msg="running hook" func=connectSources.func2
time="2021-11-16T06:47:14Z" level=info msg=closing source=default
time="2021-11-16T06:47:14Z" level=info msg="closing partition channels" source=default
time="2021-11-16T06:47:14Z" level=info msg="waiting for partition consumers to finish" source=default
time="2021-11-16T06:47:14Z" level=info msg="closing consumer" source=default
time="2021-11-16T06:47:14Z" level=info msg="failed to get pending" err="pending not available" source=default
time="2021-11-16T06:47:14Z" level=info msg="consuming Kafka logs" source=default
time="2021-11-16T06:47:14Z" level=info msg="starting poll loop" source=default
fatal error: unexpected signal during runtime execution
[signal SIGSEGV: segmentation violation code=0x1 addr=0x8 pc=0x19dc3de]

runtime stack:
runtime.throw({0x20bfc74, 0x1388})
	/usr/local/go/src/runtime/panic.go:1198 +0x71
runtime.sigpanic()
	/usr/local/go/src/runtime/signal_unix.go:719 +0x396

goroutine 141 [syscall]:
runtime.cgocall(0x19423e0, 0xc001f419d0)
	/usr/local/go/src/runtime/cgocall.go:156 +0x5c fp=0xc001f419a8 sp=0xc001f41970 pc=0x46d55c
github.com/confluentinc/confluent-kafka-go/kafka._Cfunc__rk_queue_poll(0x0, 0x1388, 0xc0001a4ce0, 0xc00057c0f0, 0x0)
	_cgo_gotypes.go:639 +0x4d fp=0xc001f419d0 sp=0xc001f419a8 pc=0x159076d
github.com/confluentinc/confluent-kafka-go/kafka.(*handle).eventPoll.func1(0x86, 0x0, 0x12, 0x1, 0x0)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/event.go:168 +0xcd fp=0xc001f41a38 sp=0xc001f419d0 pc=0x15a08cd
github.com/confluentinc/confluent-kafka-go/kafka.(*handle).eventPoll(0xc0001a2008, 0x0, 0x1388, 0x0, 0x0)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/event.go:168 +0x114 fp=0xc001f41d18 sp=0xc001f41a38 pc=0x159f3f4
github.com/confluentinc/confluent-kafka-go/kafka.(*Consumer).Poll(...)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/consumer.go:330
github.com/argoproj-labs/argo-dataflow/runner/sidecar/source/kafka.(*kafkaSource).startPollLoop(0xc00015ea80, {0x2349bd0, 0xc000524500})
	/workspace/runner/sidecar/source/kafka/kafka.go:129 +0x88 fp=0xc001f41dd8 sp=0xc001f41d18 pc=0x17f3748
github.com/argoproj-labs/argo-dataflow/runner/sidecar/source/kafka.(*kafkaSource).startPollLoop-fm({0x2349bd0, 0xc000524500})
	/workspace/runner/sidecar/source/kafka/kafka.go:125 +0x39 fp=0xc001f41e00 sp=0xc001f41dd8 pc=0x17f49d9
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext.func1()
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:185 +0x25 fp=0xc001f41e20 sp=0xc001f41e00 pc=0x1303665
k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1(0x7fb705172190)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:155 +0x67 fp=0xc001f41e60 sp=0xc001f41e20 pc=0x1303487
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0x1d81b20, {0x23091a0, 0xc00002e060}, 0x1, 0xc0002e6000)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:156 +0xb6 fp=0xc001f41ee0 sp=0xc001f41e60 pc=0x1303316
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0xc000524500, 0xb2d05e00, 0x3ff3333333333333, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:133 +0x89 fp=0xc001f41f30 sp=0xc001f41ee0 pc=0x1303209
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext({0x2349bd0, 0xc000524500}, 0xc000590f50, 0x0, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:185 +0x99 fp=0xc001f41fa0 sp=0xc001f41f30 pc=0x13035d9
github.com/argoproj-labs/argo-dataflow/runner/sidecar/source/kafka.New·dwrap·2()
	/workspace/runner/sidecar/source/kafka/kafka.go:93 +0x3b fp=0xc001f41fe0 sp=0xc001f41fa0 pc=0x17f2cfb
runtime.goexit()
	/usr/local/go/src/runtime/asm_amd64.s:1581 +0x1 fp=0xc001f41fe8 sp=0xc001f41fe0 pc=0x4d3261
created by github.com/argoproj-labs/argo-dataflow/runner/sidecar/source/kafka.New
	/workspace/runner/sidecar/source/kafka/kafka.go:93 +0xc10

goroutine 1 [runnable]:
github.com/sirupsen/logrus.Level.String(...)
	/go/pkg/mod/github.com/sirupsen/[email protected]/logrus.go:18
github.com/weaveworks/promrus.(*PrometheusHook).Fire(0xc000186060, 0x47a8f4)
	/go/pkg/mod/github.com/weaveworks/[email protected]/promrus.go:53 +0x47
github.com/sirupsen/logrus.LevelHooks.Fire(0xc000155608, 0x1555d8, 0x5)
	/go/pkg/mod/github.com/sirupsen/[email protected]/hooks.go:28 +0x82
github.com/sirupsen/logrus.(*Entry).fireHooks(0xc0002b84d0)
	/go/pkg/mod/github.com/sirupsen/[email protected]/entry.go:272 +0x1fc
github.com/sirupsen/logrus.(*Entry).log(0xc000155838, 0x4, {0xc0000f7e48, 0x11})
	/go/pkg/mod/github.com/sirupsen/[email protected]/entry.go:241 +0x191
github.com/sirupsen/logrus.(*Entry).Log(0xc0002b8460, 0x4, {0xc000155838, 0xc000155848, 0x474ae7})
	/go/pkg/mod/github.com/sirupsen/[email protected]/entry.go:293 +0x4f
github.com/sirupsen/logrus.(*Entry).Info(...)
	/go/pkg/mod/github.com/sirupsen/[email protected]/entry.go:310
github.com/bombsimon/logrusr.(*logrusr).Info(0xc0001b02c0, {0x208ef86, 0x11}, {0xc0005242c0, 0x201b5e0, 0xc00027c340})
	/go/pkg/mod/github.com/bombsimon/[email protected]/logrusr.go:111 +0x110
github.com/argoproj-labs/argo-dataflow/runner/sidecar.runHooks({0xc000590f60, 0x2, 0x8})
	/workspace/runner/sidecar/lifecycle.go:57 +0x18f
github.com/argoproj-labs/argo-dataflow/runner/sidecar.preStop({0x207eeb3, 0x5})
	/workspace/runner/sidecar/lifecycle.go:31 +0x131
github.com/argoproj-labs/argo-dataflow/runner/sidecar.Exec({0x2349bd0, 0xc000524500})
	/workspace/runner/sidecar/sidecar.go:200 +0x136b
main.main.func2(0xc00046df50, {0x2349bd0, 0xc000524500})
	/workspace/runner/main.go:98 +0x557
main.main()
	/workspace/runner/main.go:102 +0xd9

goroutine 21 [chan receive]:
k8s.io/klog/v2.(*loggingT).flushDaemon(0x0)
	/go/pkg/mod/k8s.io/klog/[email protected]/klog.go:1169 +0x6a
created by k8s.io/klog/v2.init.0
	/go/pkg/mod/k8s.io/klog/[email protected]/klog.go:417 +0xfb

goroutine 82 [syscall]:
os/signal.signal_recv()
	/usr/local/go/src/runtime/sigqueue.go:169 +0x98
os/signal.loop()
	/usr/local/go/src/os/signal/signal_unix.go:24 +0x19
created by os/signal.Notify.func1.1
	/usr/local/go/src/os/signal/signal.go:151 +0x2c

goroutine 83 [select]:
os/signal.NotifyContext.func1()
	/usr/local/go/src/os/signal/signal.go:288 +0x76
created by os/signal.NotifyContext
	/usr/local/go/src/os/signal/signal.go:287 +0x169

goroutine 99 [syscall]:
github.com/confluentinc/confluent-kafka-go/kafka._Cfunc_rd_kafka_queue_poll(0x7fb6c8000c20, 0x64)
	_cgo_gotypes.go:2382 +0x4d
github.com/confluentinc/confluent-kafka-go/kafka.(*handle).pollLogEvents.func1(0x0, 0x64)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/log.go:57 +0x51
github.com/confluentinc/confluent-kafka-go/kafka.(*handle).pollLogEvents(0xc0001ae1a0, 0xc000554060, 0x91958a, 0xc0002e6060)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/log.go:57 +0x6e
github.com/confluentinc/confluent-kafka-go/kafka.(*handle).setupLogQueue.func3()
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/handle.go:175 +0x2f
created by github.com/confluentinc/confluent-kafka-go/kafka.(*handle).setupLogQueue
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/handle.go:174 +0xf2

goroutine 100 [syscall]:
github.com/confluentinc/confluent-kafka-go/kafka._Cfunc__rk_queue_poll(0x7fb6fc000b60, 0x64, 0xc0001a4a80, 0xc00057c0c0, 0x0)
	_cgo_gotypes.go:639 +0x4d
github.com/confluentinc/confluent-kafka-go/kafka.(*handle).eventPoll.func1(0xc0001ecd20, 0xc000097f88, 0xc0001a40f8, 0xc0003118f0, 0xc000097cf8)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/event.go:168 +0xcd
github.com/confluentinc/confluent-kafka-go/kafka.(*handle).eventPoll(0xc0001ecd20, 0xc0004fd0e0, 0x64, 0x0, 0xc0002e6060)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/event.go:168 +0x114
github.com/confluentinc/confluent-kafka-go/kafka.poller(0xc0001ecd10, 0x0)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/producer.go:627 +0x52
github.com/confluentinc/confluent-kafka-go/kafka.NewProducer.func5()
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/producer.go:533 +0x29
created by github.com/confluentinc/confluent-kafka-go/kafka.NewProducer
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/producer.go:532 +0x707

goroutine 16 [IO wait]:
internal/poll.runtime_pollWait(0x7fb7065d8e18, 0x72)
	/usr/local/go/src/runtime/netpoll.go:229 +0x89
internal/poll.(*pollDesc).wait(0xc000136680, 0xc0001b6000, 0x0)
	/usr/local/go/src/internal/poll/fd_poll_runtime.go:84 +0x32
internal/poll.(*pollDesc).waitRead(...)
	/usr/local/go/src/internal/poll/fd_poll_runtime.go:89
internal/poll.(*FD).Read(0xc000136680, {0xc0001b6000, 0x94f, 0x94f})
	/usr/local/go/src/internal/poll/fd_unix.go:167 +0x25a
net.(*netFD).Read(0xc000136680, {0xc0001b6000, 0xc0001b6050, 0x12e})
	/usr/local/go/src/net/fd_posix.go:56 +0x29
net.(*conn).Read(0xc000012020, {0xc0001b6000, 0x481146, 0xc00012d7f0})
	/usr/local/go/src/net/net.go:183 +0x45
crypto/tls.(*atLeastReader).Read(0xc000532660, {0xc0001b6000, 0x0, 0x473d8d})
	/usr/local/go/src/crypto/tls/conn.go:777 +0x3d
bytes.(*Buffer).ReadFrom(0xc00018a5f8, {0x2305ec0, 0xc000532660})
	/usr/local/go/src/bytes/buffer.go:204 +0x98
crypto/tls.(*Conn).readFromUntil(0xc00018a380, {0x2309620, 0xc000012020}, 0x904)
	/usr/local/go/src/crypto/tls/conn.go:799 +0xe5
crypto/tls.(*Conn).readRecordOrCCS(0xc00018a380, 0x0)
	/usr/local/go/src/crypto/tls/conn.go:606 +0x112
crypto/tls.(*Conn).readRecord(...)
	/usr/local/go/src/crypto/tls/conn.go:574
crypto/tls.(*Conn).Read(0xc00018a380, {0xc000255000, 0x1000, 0xc000554720})
	/usr/local/go/src/crypto/tls/conn.go:1277 +0x16f
bufio.(*Reader).Read(0xc00032c6c0, {0xc0002962d8, 0x9, 0xc00038f260})
	/usr/local/go/src/bufio/bufio.go:227 +0x1b4
io.ReadAtLeast({0x2305c80, 0xc00032c6c0}, {0xc0002962d8, 0x9, 0x9}, 0x9)
	/usr/local/go/src/io/io.go:328 +0x9a
io.ReadFull(...)
	/usr/local/go/src/io/io.go:347
golang.org/x/net/http2.readFrameHeader({0xc0002962d8, 0x9, 0xc000300000}, {0x2305c80, 0xc00032c6c0})
	/go/pkg/mod/golang.org/x/[email protected]/http2/frame.go:237 +0x6e
golang.org/x/net/http2.(*Framer).ReadFrame(0xc0002962a0)
	/go/pkg/mod/golang.org/x/[email protected]/http2/frame.go:492 +0x95
golang.org/x/net/http2.(*clientConnReadLoop).run(0xc00012dfa0)
	/go/pkg/mod/golang.org/x/[email protected]/http2/transport.go:1816 +0x165
golang.org/x/net/http2.(*ClientConn).readLoop(0xc00043e480)
	/go/pkg/mod/golang.org/x/[email protected]/http2/transport.go:1738 +0x79
created by golang.org/x/net/http2.(*Transport).newClientConn
	/go/pkg/mod/golang.org/x/[email protected]/http2/transport.go:694 +0xb45

goroutine 101 [chan receive]:
github.com/confluentinc/confluent-kafka-go/kafka.channelProducer(0xc0001ecd10)
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/producer.go:556 +0x45
github.com/confluentinc/confluent-kafka-go/kafka.NewProducer.func6()
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/producer.go:547 +0x26
created by github.com/confluentinc/confluent-kafka-go/kafka.NewProducer
	/go/pkg/mod/github.com/confluentinc/[email protected]/kafka/producer.go:546 +0x785

goroutine 102 [chan receive]:
github.com/argoproj-labs/argo-dataflow/runner/sidecar/sink/kafka.New.func1({0x0, 0x0})
	/workspace/runner/sidecar/sink/kafka/kafka.go:71 +0xdf
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext.func1()
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:185 +0x25
k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1(0x7fb72f1f16d0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:155 +0x67
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0x1d81b20, {0x23091a0, 0xc00052e000}, 0x1, 0xc0002e6000)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:156 +0xb6
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0xc000524500, 0xb2d05e00, 0x3ff3333333333333, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:133 +0x89
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext({0x2349bd0, 0xc000524500}, 0xc0000ea180, 0x0, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:185 +0x99
created by github.com/argoproj-labs/argo-dataflow/runner/sidecar/sink/kafka.New
	/workspace/runner/sidecar/sink/kafka/kafka.go:69 +0x689

goroutine 103 [chan receive]:
github.com/argoproj-labs/argo-dataflow/runner/sidecar/sink/kafka.New.func2({0x0, 0x0})
	/workspace/runner/sidecar/sink/kafka/kafka.go:78 +0xa5
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext.func1()
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:185 +0x25
k8s.io/apimachinery/pkg/util/wait.BackoffUntil.func1(0x7fb7065b5db8)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:155 +0x67
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0x1d81b20, {0x23091a0, 0xc0001644b0}, 0x1, 0xc0002e6000)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:156 +0xb6
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0xc000524500, 0x3b9aca00, 0x3ff3333333333333, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:133 +0x89
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext({0x2349bd0, 0xc000524500}, 0xc000064100, 0x0, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:185 +0x99
created by github.com/argoproj-labs/argo-dataflow/runner/sidecar/sink/kafka.New
	/workspace/runner/sidecar/sink/kafka/kafka.go:76 +0x7c8

goroutine 110 [syscall]:
syscall.Syscall6(0x101, 0xffffffffffffff9c, 0xc000062160, 0x80000, 0x0, 0x0, 0x0)
	/usr/local/go/src/syscall/asm_linux_amd64.s:43 +0x5
syscall.openat(0x0, {0x209e18a, 0x0}, 0x0, 0x0)
	/usr/local/go/src/syscall/zsyscall_linux_amd64.go:69 +0x105
syscall.Open(...)
	/usr/local/go/src/syscall/syscall_linux.go:155
os.openFileNolog({0x209e18a, 0x0}, 0x0, 0x2000000)
	/usr/local/go/src/os/file_unix.go:217 +0x9b
os.OpenFile({0x209e18a, 0x1a}, 0x0, 0x0)
	/usr/local/go/src/os/file.go:338 +0x45
github.com/argoproj-labs/argo-dataflow/runner/sidecar.connectOutFIFO.func1.1(0xc0000ea1e0, {0x2349bd0, 0xc000524500})
	/workspace/runner/sidecar/out.go:75 +0x59
github.com/argoproj-labs/argo-dataflow/runner/sidecar.connectOutFIFO.func1()
	/workspace/runner/sidecar/out.go:104 +0x70
created by github.com/argoproj-labs/argo-dataflow/runner/sidecar.connectOutFIFO
	/workspace/runner/sidecar/out.go:72 +0xc6

goroutine 111 [IO wait]:
internal/poll.runtime_pollWait(0x7fb7065d8b60, 0x72)
	/usr/local/go/src/runtime/netpoll.go:229 +0x89
internal/poll.(*pollDesc).wait(0xc0001b2200, 0x2, 0x0)
	/usr/local/go/src/internal/poll/fd_poll_runtime.go:84 +0x32
internal/poll.(*pollDesc).waitRead(...)
	/usr/local/go/src/internal/poll/fd_poll_runtime.go:89
internal/poll.(*FD).Accept(0xc0001b2200)
	/usr/local/go/src/internal/poll/fd_unix.go:402 +0x22c
net.(*netFD).accept(0xc0001b2200)
	/usr/local/go/src/net/fd_unix.go:173 +0x35
net.(*TCPListener).accept(0xc000532090)
	/usr/local/go/src/net/tcpsock_posix.go:140 +0x28
net.(*TCPListener).Accept(0xc000532090)
	/usr/local/go/src/net/tcpsock.go:262 +0x3d
net/http.(*Server).Serve(0xc000296000, {0x2331498, 0xc000532090})
	/usr/local/go/src/net/http/server.go:3001 +0x394
net/http.(*Server).ListenAndServe(0xc000296000)
	/usr/local/go/src/net/http/server.go:2930 +0x7d
github.com/argoproj-labs/argo-dataflow/runner/sidecar.Exec.func9()
	/workspace/runner/sidecar/sidecar.go:175 +0xae
created by github.com/argoproj-labs/argo-dataflow/runner/sidecar.Exec
	/workspace/runner/sidecar/sidecar.go:172 +0x10e6

goroutine 112 [IO wait]:
internal/poll.runtime_pollWait(0x7fb7065d8c48, 0x72)
	/usr/local/go/src/runtime/netpoll.go:229 +0x89
internal/poll.(*pollDesc).wait(0xc000320080, 0x10006d800, 0x0)
	/usr/local/go/src/internal/poll/fd_poll_runtime.go:84 +0x32
internal/poll.(*pollDesc).waitRead(...)
	/usr/local/go/src/internal/poll/fd_poll_runtime.go:89
internal/poll.(*FD).Accept(0xc000320080)
	/usr/local/go/src/internal/poll/fd_unix.go:402 +0x22c
net.(*netFD).accept(0xc000320080)
	/usr/local/go/src/net/fd_unix.go:173 +0x35
net.(*TCPListener).accept(0xc0000e4150)
	/usr/local/go/src/net/tcpsock_posix.go:140 +0x28
net.(*TCPListener).Accept(0xc0000e4150)
	/usr/local/go/src/net/tcpsock.go:262 +0x3d
crypto/tls.(*listener).Accept(0xc0000e4168)
	/usr/local/go/src/crypto/tls/tls.go:66 +0x2d
net/http.(*Server).Serve(0xc000296380, {0x2325288, 0xc0000e4168})
	/usr/local/go/src/net/http/server.go:3001 +0x394
net/http.(*Server).ServeTLS(0xc000296380, {0x2331498, 0xc0000e4150}, {0x0, 0xc0015c7f38}, {0x0, 0x0})
	/usr/local/go/src/net/http/server.go:3073 +0x3fe
net/http.(*Server).ListenAndServeTLS(0xc000296380, {0x0, 0x0}, {0x0, 0x0})
	/usr/local/go/src/net/http/server.go:3228 +0x12f
github.com/argoproj-labs/argo-dataflow/runner/sidecar.Exec.func11()
	/workspace/runner/sidecar/sidecar.go:188 +0xb8
created by github.com/argoproj-labs/argo-dataflow/runner/sidecar.Exec
	/workspace/runner/sidecar/sidecar.go:185 +0x130f

goroutine 84 [IO wait]:
internal/poll.runtime_pollWait(0x7fb7065d8d30, 0x72)
	/usr/local/go/src/runtime/netpoll.go:229 +0x89
internal/poll.(*pollDesc).wait(0xc000136300, 0xc000299000, 0x0)
	/usr/local/go/src/internal/poll/fd_poll_runtime.go:84 +0x32
internal/poll.(*pollDesc).waitRead(...)
	/usr/local/go/src/internal/poll/fd_poll_runtime.go:89
internal/poll.(*FD).Read(0xc000136300, {0xc000299000, 0x1000, 0x1000})
	/usr/local/go/src/internal/poll/fd_unix.go:167 +0x25a
net.(*netFD).Read(0xc000136300, {0xc000299000, 0x1, 0x0})
	/usr/local/go/src/net/fd_posix.go:56 +0x29
net.(*conn).Read(0xc0000e8018, {0xc000299000, 0xe, 0xc0000d0600})
	/usr/local/go/src/net/net.go:183 +0x45
net/http.(*persistConn).Read(0xc00025c240, {0xc000299000, 0x4b2dc0, 0xc0001f4ec8})
	/usr/local/go/src/net/http/transport.go:1926 +0x4e
bufio.(*Reader).fill(0xc0002d20c0)
	/usr/local/go/src/bufio/bufio.go:101 +0x103
bufio.(*Reader).Peek(0xc0002d20c0, 0x1)
	/usr/local/go/src/bufio/bufio.go:139 +0x5d
net/http.(*persistConn).readLoop(0xc00025c240)
	/usr/local/go/src/net/http/transport.go:2087 +0x1ac
created by net/http.(*Transport).dialConn
	/usr/local/go/src/net/http/transport.go:1747 +0x1e05

goroutine 85 [select]:
net/http.(*persistConn).writeLoop(0xc00025c240)
	/usr/local/go/src/net/http/transport.go:2386 +0xfb
created by net/http.(*Transport).dialConn
	/usr/local/go/src/net/http/transport.go:1748 +0x1e65

goroutine 140 [select]:
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0x1d81b20, {0x23091a0, 0xc000311620}, 0x1, 0xc0002e6000)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:167 +0x13b
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0xc000524500, 0xb2d05e00, 0x3ff3333333333333, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:133 +0x89
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext({0x2349bd0, 0xc000524500}, 0xc0001ae920, 0x0, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:185 +0x99
created by github.com/argoproj-labs/argo-dataflow/runner/sidecar/source/kafka.New
	/workspace/runner/sidecar/source/kafka/kafka.go:68 +0x8d3

goroutine 142 [select]:
k8s.io/apimachinery/pkg/util/wait.BackoffUntil(0x1d81b20, {0x23091a0, 0xc0002c8060}, 0x1, 0xc0002e6000)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:167 +0x13b
k8s.io/apimachinery/pkg/util/wait.JitterUntil(0xc000524500, 0x37e11d600, 0x3ff3333333333333, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:133 +0x89
k8s.io/apimachinery/pkg/util/wait.JitterUntilWithContext({0x2349bd0, 0xc000524500}, 0xc000557f50, 0x0, 0x0, 0x0)
	/go/pkg/mod/k8s.io/[email protected]/pkg/util/wait/wait.go:185 +0x99
created by github.com/argoproj-labs/argo-dataflow/runner/sidecar.connectSources
	/workspace/runner/sidecar/sources.go:197 +0x1c4c

Pipeline:

bug

opened by Udit107710 7

feat: Adding Python SDK for git step with Python3-9 runtime.
This SDK is implementing the image contract(https://github.com/argoproj-labs/argo-dataflow/blob/main/docs/IMAGE_CONTRACT.md), by:

Responding to /ready with a 204 status code once it is able to start serving requests,

Accepting POST request to /messages, and responding with 201, 204 or 500 accordingly,

Handles SIGTERM gracefully, giving server time to finish processing the request before terminating.
opened by domderen 7
Dead Letter Queues failing

cross-posting

Hey, I see that v0.3.0 has support for Dead letter queues. I just gave it a spin as per this comment from the PR - but got stuck at a few places

Context: I have a kafka source and a sink for the first step in my pipeline and the messages from the sink will be consumed by the subsequent step.

Problem: I wanted to know if the failed messages in the first step can be replayed through a DLQ? if so how do I add the failed messages to a DLQ? (I tried adding another kafka sink and marking it as a DLQ, didn't work as expected)

If I keep the same name for both sinks, it errs out.

bug

opened by inishchith 6
Step not scaling

The above command was for scaling a Container Step The command succeeds but Step didn't scale I tried scaling Code Step, but the same result: command doesn't return any error but scaling does not happen
bug no-issue-activity more-information-needed

opened by Udit107710 6
Quickstart does not seem to work with examples

I am trying to play around with dataflow to understand it a bit better but the quick start deployment doesnt seem to work correctly. The input pod that is created which seems like its supposed to be pushing messages to kafka fails. It looks like some other pod is missing. The input pod is using curl to hit a testapi, but there isnt anything like that deployed in any of the quick start instructions or manifests. Am I missing something or is there more to the setup that is missing from the instructions?
no-issue-activity more-information-needed

opened by Rob-Fletcher 6
database source
As a user, I'd like to process records from a database.

How do I define the query?

Do I need to action new records? I.e. each new record is a new message.

See #146
no-issue-activity
opened by alexec 6
database sink
As a user I'd like to be able to sink messages to a database. Open questions:

Sink to a single table presumably?

How do we define the statement to execute?

Inserts only? What about primary key conflicts?

See #147
no-issue-activity
opened by alexec 6
build(deps): bump sigs.k8s.io/controller-runtime from 0.11.2 to 0.14.1
Bumps sigs.k8s.io/controller-runtime from 0.11.2 to 0.14.1.

Release notes

Sourced from sigs.k8s.io/controller-runtime's releases.

v0.14.1

Changes since v0.14.0

:bug: Bug Fixes

🐛 controllers without For() fail to start by @k8s-infra-cherrypick-robot in kubernetes-sigs/controller-runtime#2108

Full Changelog: https://github.com/kubernetes-sigs/controller-runtime/compare/v0.14.0...v0.14.1

v0.14.0

Changes since v0.13.1

:warning: Breaking Changes

Add Get functionality to SubResourceClient (#2094)

Allow configuring RecoverPanic for controllers globally (#2093)

Add client.SubResourceWriter (#2072)

Support registration and removal of event handler (#2046)

Update Kubernetes dependencies to v0.26 (#2043, #2087)

Zap log: Default to RFC3339 time encoding (#2029)

cache.BuilderWithOptions inherit options from caller (#1980)

:sparkles: New Features

Builder: Do not require For (#2091)

support disable deepcopy on list funcion (#2076)

Add cluster.NewClientFunc with options (#2054)

Tidy up startup logging of kindWithCache source (#2057)

Add function to get reconcileID from context (#2056)

feat: add NOT predicate (#2031)

Allow to provide a custom lock interface to manager (#2027)

Add tls options to manager.Options (#2023)

Update Go version to 1.19 (#1986)

:bug: Bug Fixes

Prevent manager from getting started a second time (#2090)

Missing error log for in-cluster config (#2051)

Skip custom mutation handler when delete a CR (#2049)

fix: improve semantics of combining cache selectorsByObject (#2039)

Conversion webhook should not panic when conversion request is nil (#1970)

:seedling: Others

Prepare for release 0.14 (#2100)

Generate files and update modules (#2096)

Bump github.com/onsi/ginkgo/v2 from 2.5.1 to 2.6.0 (#2097)

Bump golang.org/x/time (#2089)

Update OWNERS: remove inactive members, promote fillzpp sbueringer (#2088, #2092)

Default ENVTEST version to a working one (1.24.2) (#2081)

Update golangci-lint to v1.50.1 (#2080)

Bump go.uber.org/zap from 1.23.0 to 1.24.0 (#2077)

Bump golang.org/x/sys from 0.2.0 to 0.3.0 (#2078)

Ignore Kubernetes Dependencies in Dependabot (#2071)

... (truncated)

Commits

84c5c9f 🐛 controllers without For() fail to start (#2108)

ddcb99d Merge pull request #2100 from vincepri/release-0.14

69f0938 Merge pull request #2094 from alvaroaleman/subresoruce-get

8738e91 Merge pull request #2091 from alvaroaleman/no-for

ca4b4de Merge pull request #2096 from lucacome/generate

5673341 Merge pull request #2097 from kubernetes-sigs/dependabot/go_modules/github.co...

7333aed :seedling: Bump github.com/onsi/ginkgo/v2 from 2.5.1 to 2.6.0

d4f1e82 Generate files and update modules

a387bf4 Merge pull request #2093 from alvaroaleman/recover-panic-globally

da7dd5d :warning: Allow configuring RecoverPanic for controllers globally

Additional commits viewable in compare view

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

Dependabot commands and options

You can trigger Dependabot actions by commenting on this PR:

@dependabot rebase will rebase this PR

@dependabot recreate will recreate this PR, overwriting any edits that have been made to it

@dependabot merge will merge this PR after your CI passes on it

@dependabot squash and merge will squash and merge this PR after your CI passes on it

@dependabot cancel merge will cancel a previously requested merge and block automerging

@dependabot reopen will reopen this PR if it is closed

@dependabot close will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually

@dependabot ignore this major version will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)

@dependabot ignore this minor version will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)

@dependabot ignore this dependency will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)

dependencies go
opened by dependabot[bot] 0
build(deps): bump github.com/onsi/gomega from 1.20.1 to 1.24.2
Bumps github.com/onsi/gomega from 1.20.1 to 1.24.2.

Release notes

Sourced from github.com/onsi/gomega's releases.

v1.24.2

1.24.2

Fixes

Correctly handle assertion failure panics for eventually/consistnetly "g Gomega"s in a goroutine [78f1660]

docs:Fix typo "you an" -> "you can" (#607) [3187c1f]

fixes issue #600 (#606) [808d192]

Maintenance

Bump golang.org/x/net from 0.2.0 to 0.4.0 (#611) [6ebc0bf]

Bump nokogiri from 1.13.9 to 1.13.10 in /docs (#612) [258cfc8]

Bump github.com/onsi/ginkgo/v2 from 2.5.0 to 2.5.1 (#609) [e6c3eb9]

v1.24.1

No release notes provided.

v1.24.0

1.24.0

Features

Introducting gcustom - a convenient mechanism for building custom matchers.

This is an RC release for gcustom. The external API may be tweaked in response to feedback however it is expected to remain mostly stable.

Maintenance

Update BeComparableTo documentation [756eaa0]

v1.23.0

1.23.0

Features

Custom formatting on a per-type basis can be provided using format.RegisterCustomFormatter() -- see the docs here

Substantial improvement have been made to StopTrying():

Users can now use StopTrying().Wrap(err) to wrap errors and StopTrying().Attach(description, object) to attach arbitrary objects to the StopTrying() error

StopTrying() is now always interpreted as a failure. If you are an early adopter of StopTrying() you may need to change your code as the prior version would match against the returned value even if StopTrying() was returned. Going forward the StopTrying() api should remain stable.

StopTrying() and StopTrying().Now() can both be used in matchers - not just polled functions.

TryAgainAfter(duration) is used like StopTrying() but instructs Eventually and Consistently that the poll should be tried again after the specified duration. This allows you to dynamically adjust the polling duration.

ctx can now be passed-in as the first argument to Eventually and Consistently.

Maintenance

Bump github.com/onsi/ginkgo/v2 from 2.3.0 to 2.3.1 (#597) [afed901]

Bump nokogiri from 1.13.8 to 1.13.9 in /docs (#599) [7c691b3]

Bump github.com/google/go-cmp from 0.5.8 to 0.5.9 (#587) [ff22665]

... (truncated)

Changelog

Sourced from github.com/onsi/gomega's changelog.

1.24.2

Fixes

Correctly handle assertion failure panics for eventually/consistnetly "g Gomega"s in a goroutine [78f1660]

docs:Fix typo "you an" -> "you can" (#607) [3187c1f]

fixes issue #600 (#606) [808d192]

Maintenance

Bump golang.org/x/net from 0.2.0 to 0.4.0 (#611) [6ebc0bf]

Bump nokogiri from 1.13.9 to 1.13.10 in /docs (#612) [258cfc8]

Bump github.com/onsi/ginkgo/v2 from 2.5.0 to 2.5.1 (#609) [e6c3eb9]

1.24.1

Fixes

maintain backward compatibility for Eventually and Consisntetly's signatures [4c7df5e]

fix small typo (#601) [ea0ebe6]

Maintenance

Bump golang.org/x/net from 0.1.0 to 0.2.0 (#603) [1ba8372]

Bump github.com/onsi/ginkgo/v2 from 2.4.0 to 2.5.0 (#602) [f9426cb]

fix label-filter in test.yml [d795db6]

stop running flakey tests and rely on external network dependencies in CI [7133290]

1.24.0

Features

Introducting gcustom - a convenient mechanism for building custom matchers.

This is an RC release for gcustom. The external API may be tweaked in response to feedback however it is expected to remain mostly stable.

Maintenance

Update BeComparableTo documentation [756eaa0]

1.23.0

Features

Custom formatting on a per-type basis can be provided using format.RegisterCustomFormatter() -- see the docs here

Substantial improvement have been made to StopTrying():

Users can now use StopTrying().Wrap(err) to wrap errors and StopTrying().Attach(description, object) to attach arbitrary objects to the StopTrying() error

StopTrying() is now always interpreted as a failure. If you are an early adopter of StopTrying() you may need to change your code as the prior version would match against the returned value even if StopTrying() was returned. Going forward the StopTrying() api should remain stable.

StopTrying() and StopTrying().Now() can both be used in matchers - not just polled functions.

TryAgainAfter(duration) is used like StopTrying() but instructs Eventually and Consistently that the poll should be tried again after the specified duration. This allows you to dynamically adjust the polling duration.

ctx can now be passed-in as the first argument to Eventually and Consistently.

... (truncated)

Commits

c3aef8b v1.24.2

78f1660 Correctly handle assertion failure panics for eventually/consistnetly "g Gom...

6ebc0bf Bump golang.org/x/net from 0.2.0 to 0.4.0 (#611)

258cfc8 Bump nokogiri from 1.13.9 to 1.13.10 in /docs (#612)

e6c3eb9 Bump github.com/onsi/ginkgo/v2 from 2.5.0 to 2.5.1 (#609)

3187c1f docs:Fix typo "you an" -> "you can" (#607)

808d192 fixes issue #600 (#606)

3eef0d7 v1.24.1

4c7df5e maintain backward compatibility for Eventually and Consisntetly's signatures

1ba8372 Bump golang.org/x/net from 0.1.0 to 0.2.0 (#603)

Additional commits viewable in compare view

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

Dependabot commands and options

You can trigger Dependabot actions by commenting on this PR:

@dependabot rebase will rebase this PR

@dependabot recreate will recreate this PR, overwriting any edits that have been made to it

@dependabot merge will merge this PR after your CI passes on it

@dependabot squash and merge will squash and merge this PR after your CI passes on it

@dependabot cancel merge will cancel a previously requested merge and block automerging

@dependabot reopen will reopen this PR if it is closed

@dependabot close will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually

@dependabot ignore this major version will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)

@dependabot ignore this minor version will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)

@dependabot ignore this dependency will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)

dependencies go
opened by dependabot[bot] 0