diff --git a/.github/workflows/build-push.yaml b/.github/workflows/build-push.yaml index 5e018bbe..b2fbe938 100644 --- a/.github/workflows/build-push.yaml +++ b/.github/workflows/build-push.yaml @@ -22,7 +22,8 @@ jobs: "pkg/reducer/examples/counter", "pkg/reducer/examples/sum", "pkg/reducestreamer/examples/counter", "pkg/reducestreamer/examples/sum", "pkg/sessionreducer/examples/counter", "pkg/sessionreducer/examples/sum", "pkg/sideinput/examples/simple-sideinput/udf", "pkg/sideinput/examples/simple-sideinput", "pkg/sinker/examples/log", - "pkg/sourcer/examples/simple_source", "pkg/sourcetransformer/examples/assign_event_time", "pkg/sourcetransformer/examples/event_time_filter" + "pkg/sourcer/examples/simple_source", "pkg/sourcetransformer/examples/assign_event_time", "pkg/sourcetransformer/examples/event_time_filter", + "pkg/sinker/examples/fallback" ] steps: diff --git a/pkg/sinker/examples/fallback/Dockerfile b/pkg/sinker/examples/fallback/Dockerfile new file mode 100644 index 00000000..b5c9c889 --- /dev/null +++ b/pkg/sinker/examples/fallback/Dockerfile @@ -0,0 +1,20 @@ +#################################################################################################### +# base +#################################################################################################### +FROM alpine:3.12.3 as base +RUN apk update && apk upgrade && \ + apk add ca-certificates && \ + apk --no-cache add tzdata + +COPY dist/fallback-example /bin/fallback-example +RUN chmod +x /bin/fallback-example + +#################################################################################################### +# fallback +#################################################################################################### +FROM scratch as fallback +ARG ARCH +COPY --from=base /usr/share/zoneinfo /usr/share/zoneinfo +COPY --from=base /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt +COPY --from=base /bin/fallback-example /bin/fallback-example +ENTRYPOINT [ "/bin/fallback-example" ] diff --git a/pkg/sinker/examples/fallback/Makefile b/pkg/sinker/examples/fallback/Makefile new file mode 100644 index 00000000..2d7a7445 --- /dev/null +++ b/pkg/sinker/examples/fallback/Makefile @@ -0,0 +1,18 @@ +TAG ?= stable +PUSH ?= false + +.PHONY: build +build: + CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -v -o ./dist/log-example main.go + +.PHONY: image-push +image-push: build + docker buildx build -t "quay.io/numaio/numaflow-go/fb-sink-log:${TAG}" --platform linux/amd64,linux/arm64 --target log . --push + +.PHONY: image +image: build + docker build -t "quay.io/numaio/numaflow-go/fb-sink-log:${TAG}" --target log . + @if [ "$(PUSH)" = "true" ]; then docker push "quay.io/numaio/numaflow-go/fb-sink-log:${TAG}"; fi + +clean: + -rm -rf ./dist diff --git a/pkg/sinker/examples/fallback/README.md b/pkg/sinker/examples/fallback/README.md new file mode 100644 index 00000000..5399c12e --- /dev/null +++ b/pkg/sinker/examples/fallback/README.md @@ -0,0 +1,3 @@ +# Fallback Sink Example + +An example User Defined Sink that returns a fallback response so that the payload will be sent to the fallback sink. \ No newline at end of file diff --git a/pkg/sinker/examples/fallback/go.mod b/pkg/sinker/examples/fallback/go.mod new file mode 100644 index 00000000..68c596f1 --- /dev/null +++ b/pkg/sinker/examples/fallback/go.mod @@ -0,0 +1,17 @@ +module log_sink + +go 1.20 + +replace github.com/numaproj/numaflow-go => ../../../.. + +require github.com/numaproj/numaflow-go v0.7.0-rc2 + +require ( + github.com/golang/protobuf v1.5.3 // indirect + golang.org/x/net v0.9.0 // indirect + golang.org/x/sys v0.7.0 // indirect + golang.org/x/text v0.9.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect + google.golang.org/grpc v1.57.0 // indirect + google.golang.org/protobuf v1.31.0 // indirect +) diff --git a/pkg/sinker/examples/fallback/go.sum b/pkg/sinker/examples/fallback/go.sum new file mode 100644 index 00000000..95c8479a --- /dev/null +++ b/pkg/sinker/examples/fallback/go.sum @@ -0,0 +1,24 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= +github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM= +golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns= +golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU= +golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE= +golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 h1:0nDDozoAU19Qb2HwhXadU8OcsiO/09cnTqhUtq2MEOM= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA= +google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw= +google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8= +google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/pkg/sinker/examples/fallback/main.go b/pkg/sinker/examples/fallback/main.go new file mode 100644 index 00000000..9ddb27a5 --- /dev/null +++ b/pkg/sinker/examples/fallback/main.go @@ -0,0 +1,33 @@ +package main + +import ( + "context" + "fmt" + "log" + + sinksdk "github.com/numaproj/numaflow-go/pkg/sinker" +) + +// fbLogSink is a sinker implementation that logs the input to stdout +type fbLogSink struct { +} + +func (l *fbLogSink) Sink(ctx context.Context, datumStreamCh <-chan sinksdk.Datum) sinksdk.Responses { + result := sinksdk.ResponsesBuilder() + for d := range datumStreamCh { + _ = d.EventTime() + _ = d.Watermark() + id := d.ID() + fmt.Println("Primary sink under maintenance, writing to fallback sink - ", string(d.Value())) + // write to fallback sink + result = result.Append(sinksdk.ResponseFallback(id)) + } + return result +} + +func main() { + err := sinksdk.NewServer(&fbLogSink{}).Start(context.Background()) + if err != nil { + log.Panic("Failed to start sink function server: ", err) + } +}