WIP - V2 alpha

.dockerignore

@@ -0,0 +1,5 @@
+integration/**/*
+Dockerfile
+docs/**/*
+benchmark/**/*
+.vscode

.github/workflows/push.yaml

@@ -0,0 +1,38 @@
+on: push
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: install latest buildx
+        run: mkdir -p ~/.docker/cli-plugins && wget -O ~/.docker/cli-plugins/docker-buildx https://github.com/docker/buildx/releases/download/v0.4.1/buildx-v0.4.1.linux-amd64 && chmod a+x  ~/.docker/cli-plugins/docker-buildx
+      - name: build
+        env:
+          DOCKER_BUILDKIT: 1
+          DOCKER_CLI_EXPERIMENTAL: enabled
+        working-directory: ./integration/docker
+        run: docker buildx bake -f ./docker-compose.base.yaml -f ./docker-compose.worker.yaml -f ./docker-compose.producer.yaml --load  
+      - name: test-multi
+        working-directory: ./integration/tests
+        run: ./multi.sh
+      - name: Push to GitHub Packages
+        uses: docker/build-push-action@v1
+        env:
+          DOCKER_BUILDKIT: 1
+        with:
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+          registry: docker.pkg.github.com
+          repository: soluto/dqd/app
+          tag_with_sha: ${{ github.ref == 'refs/heads/master'}}
+          tag_with_ref: ${{ github.ref != 'refs/heads/master' || startsWith(github.ref, 'refs/tags/') }}
+      - name: Push to Dockerhub
+        uses: docker/build-push-action@v1
+        env:
+          DOCKER_BUILDKIT: 1
+        with:
+          username: solutodqd
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+          repository: soluto/dqd
+          tag_with_sha: ${{ github.ref == 'refs/heads/master'}}
+          tag_with_ref: ${{ github.ref != 'refs/heads/master' || startsWith(github.ref, 'refs/tags/') }}

.gitignore

@@ -12,4 +12,5 @@
 *.out
 
 node_modules
-.vscode/launch.json
+.vscode/launch.json
+cloud-envs

Dockerfile

@@ -1,18 +1,30 @@
-# ------------- Build ----------------------
-FROM golang:1.10.1-stretch as build
+# syntax = docker/dockerfile:1.0-experimental
+FROM golang:1.14.0-alpine as builder
+RUN apk update && apk add --no-cache git ca-certificates && update-ca-certificates
 
-ENV CGO_ENABLED=0
-ENV GOOS=linux
-ENV GOBIN=$GOPATH/bin
-WORKDIR /src
+ENV UID=10001
+RUN adduser \    
+    --disabled-password \    
+    --gecos "" \    
+    --home "/nonexistent" \    
+    --shell "/sbin/nologin" \    
+    --no-create-home \    
+    --uid "${UID}" \    
+    appuser 
 
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN --mount=type=cache,target=/root/.cache/go-build go mod download 
+RUN go mod verify
 COPY . .
-RUN go get .
-RUN go build -o main .
 
-# ------------- Release ----------------------
-FROM scratch as release
+RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /go/bin/dqd
+
+FROM scratch
 
-COPY --from=build /etc/ssl/certs/ /etc/ssl/certs/
-COPY --from=build /src/main /
-CMD [ "/main" ]
+COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
+COPY --from=builder /go/bin/dqd /dqd
+COPY --from=builder /etc/passwd /etc/passwd
+COPY --from=builder /etc/group /etc/group
+USER appuser:appuser
+ENTRYPOINT ["/dqd"]

cmd/dqd.go

@@ -0,0 +1,54 @@
+package cmd
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	gofigure "github.com/NCAR/go-figure"
+	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
+)
+
+func ConfigurationError(err error) {
+	fmt.Println(err)
+	pflag.Usage()
+	os.Exit(1)
+}
+
+func Load() (*viper.Viper, error) {
+	v := viper.New()
+	configFiles := pflag.CommandLine.StringSlice("config", []string{"./dqd.yaml"}, "Load a DQD config file")
+	configDirs := pflag.CommandLine.StringSlice("configDir", []string{"/etc/dqd", "/dqd/config"}, "Lookup for config files in these folders")
+	configOverrides := pflag.CommandLine.StringSlice("set", []string{}, "Override specific configuration keys")
+	pflag.Parse()
+	v.SetConfigType("yaml")
+	v.SetEnvPrefix("dqd")
+	v.AutomaticEnv()
+	err := gofigure.Parse(v, *configDirs)
+	if err != nil {
+		panic(fmt.Errorf("Fatal error config file: %s \n", err))
+	}
+	for _, f := range *configFiles {
+		r, err := os.Open(f)
+		defer r.Close()
+		if err == nil {
+			v.MergeConfig(r)
+		} else {
+			if f != "./dqd.yaml" {
+				return nil, err
+			}
+		}
+
+	}
+
+	for _, override := range *configOverrides {
+		entry := strings.Split(override, "=")
+		if len(entry) != 2 {
+			return nil, fmt.Errorf("invalid set value '%v'", override)
+		}
+		v.Set(entry[0], entry[1])
+	}
+
+	return v, nil
+}

config/config.go

@@ -0,0 +1,173 @@
+package config
+
+import (
+	"fmt"
+
+	"github.com/soluto/dqd/handlers"
+	"github.com/soluto/dqd/listeners"
+	"github.com/soluto/dqd/pipe"
+	"github.com/soluto/dqd/providers/azure"
+	"github.com/soluto/dqd/providers/servicebus"
+	"github.com/soluto/dqd/providers/sqs"
+	"github.com/soluto/dqd/utils"
+	v1 "github.com/soluto/dqd/v1"
+	"github.com/spf13/viper"
+)
+
+type App struct {
+	Sources   map[string]*v1.Source
+	Listeners []listeners.Listener
+	Workers   []*pipe.Worker
+}
+
+var sourceProviders = map[string]struct {
+	v1.ConsumerFactory
+	v1.ProducerFactory
+}{
+	"azure-queue": {
+		&azure.AzureQueueClientFactory{},
+		&azure.AzureQueueClientFactory{},
+	},
+	"sqs": {
+		&sqs.SQSClientFactory{},
+		&sqs.SQSClientFactory{},
+	},
+	"service-bus": {
+		&servicebus.ServiceBusClientFactory{},
+		&servicebus.ServiceBusClientFactory{},
+	},
+	"io": {
+		&utils.IoSourceFactory{},
+		&utils.IoSourceFactory{},
+	},
+}
+
+func createSources(v *viper.Viper) map[string]*v1.Source {
+	sources := map[string]*v1.Source{}
+	for sourceName, subSource := range utils.ViperSubMap(v, "sources") {
+		sourceType := subSource.GetString("type")
+		factory, exist := sourceProviders[sourceType]
+		if !exist {
+			panic(fmt.Errorf("FATAL - Unkown source provider:%v", sourceType))
+		}
+		sources[sourceName] = v1.NewSource(factory, factory, subSource, sourceName)
+	}
+	return sources
+}
+
+func getSource(sources map[string]*v1.Source, sourceName string) *v1.Source {
+	source, exists := sources[sourceName]
+	if !exists {
+		panic(fmt.Sprintf("Missing source definition: %v", sourceName))
+	}
+	return source
+}
+
+func getPipeSources(sources map[string]*v1.Source, v *viper.Viper) (pipeSources []*v1.Source) {
+	sourcesConfig := v.GetStringSlice("sources")
+	for _, s := range sourcesConfig {
+		pipeSources = append(pipeSources, getSource(sources, s))
+	}
+	if len(pipeSources) == 0 {
+		pipeSources = []*v1.Source{getSource(sources, v.GetString("source"))}
+	}
+	return
+}
+
+func createHandler(v *viper.Viper) handlers.Handler {
+	if v == nil {
+		panic("no handler define for pipe, use 'none' handler if it's the desired behavior")
+	}
+	if v.Get("none") != nil {
+		return handlers.None
+	}
+	v.SetDefault("http.path", "/")
+	v.SetDefault("http.host", "localhost")
+	v.SetDefault("http.port", 80)
+
+	httpEndpoint := v.GetString("http.endpoint")
+	if httpEndpoint == "" {
+		httpEndpoint = fmt.Sprintf("http://%v:%v%v", v.GetString("http.host"), v.GetString("http.port"), v.GetString("http.path"))
+	}
+	return handlers.NewHttpHandler(httpEndpoint)
+}
+
+func createWorkers(v *viper.Viper, sources map[string]*v1.Source) []*pipe.Worker {
+	var wList []*pipe.Worker
+	pipesConfig := utils.ViperSubMap(v, "pipes")
+	for name, pipeConfig := range pipesConfig {
+		pipeConfig.SetDefault("rate.init", 10)
+		pipeConfig.SetDefault("rate.min", 1)
+		pipeConfig.SetDefault("rate.window", "30s")
+		pipeConfig.SetDefault("source", "default")
+		handler := createHandler(pipeConfig.Sub("handler"))
+
+		pipeSources := getPipeSources(sources, pipeConfig)
+
+		var opts = []pipe.WorkerOption{}
+		writeToSource := pipeConfig.GetString("onError.writeTo.source")
+		if writeToSource != "" {
+			opts = append(opts, pipe.WithErrorSource(getSource(sources, writeToSource)))
+		}
+
+		if pipeConfig.IsSet("rate.fixed") {
+			opts = append(opts, pipe.WithFixedRate(pipeConfig.GetInt("rate.fixed")))
+		} else {
+			opts = append(opts, pipe.WithDynamicRate(pipeConfig.GetInt("rate.init"),
+				pipeConfig.GetInt("rate.min"),
+				pipeConfig.GetDuration("rate.window")))
+		}
+		output := pipeConfig.GetString("output")
+		if output != "" {
+			opts = append(opts, pipe.WithOutput(getSource(sources, output)))
+		} else {
+			opts = append(opts, pipe.WithDynamicRate(pipeConfig.GetInt("rate.init"),
+				pipeConfig.GetInt("rate.min"),
+				pipeConfig.GetDuration("rate.window")))
+		}
+
+		wList = append(wList, pipe.NewWorker(
+			name,
+			pipeSources,
+			handler,
+			opts...,
+		))
+	}
+	return wList
+}
+
+func createListeners(v *viper.Viper, sources map[string]*v1.Source) []listeners.Listener {
+	v.SetDefault("listeners.http.host", "0.0.0.0:9999")
+	host := v.GetString("listeners.http.host")
+	listener := listeners.Http(host)
+	for _, s := range sources {
+		listener.Add(s, viper.New())
+	}
+	return []listeners.Listener{listener}
+}
+
+func CreateApp(v *viper.Viper) (_ *App, err error) {
+	defer func() {
+		if r := recover(); r != nil {
+			err = fmt.Errorf("%v", r)
+		}
+	}()
+
+	err = utils.NormalizeEntityConfig(v, "pipe", "pipes")
+	if err != nil {
+		return
+	}
+	err = utils.NormalizeEntityConfig(v, "source", "sources")
+	if err != nil {
+		return
+	}
+
+	sources := createSources(v)
+	listeners := createListeners(v, sources)
+	workers := createWorkers(v, sources)
+	return &App{
+		sources,
+		listeners,
+		workers,
+	}, nil
+}

docs/index.md

@@ -0,0 +1,2 @@
+# DQD
+

docs/sources/azure.md

@@ -0,0 +1,20 @@
+
+
+```yaml
+source:
+  type: azure-queue
+
+  # Location
+  storageAccount: test
+  queue: dqd
+  #connection: http://azure:10001/devstoreaccount1 useful for local testing
+
+  # Credentials
+  storageAccountKey: ****
+  sasToken: ****
+
+  # Options
+  visibilityTimeoutInSeconds: 100 # defaults to 60
+  maxDequeueCount: 1 # deaults to 5
+  retryVisiblityTimeoutInSeconds: [10, 500, 600]
+``` 

docs/sources/service-bus.md

@@ -0,0 +1,10 @@
+```yaml
+source:
+    type: service-bus
+
+    connectionString: ""
+    topic: "my-topic"
+    subscription: "my sub"
+
+    prefetchCount: 30 #defaults to 1
+```

docs/sources/sqs.md

@@ -0,0 +1,19 @@
+
+# SQS Source
+
+Credentials are taken by the aws sdk defaukt provider chain:
+https://docs.aws.amazon.com/sdk-for-go/v1/developer-guide/configuring-sdk.html 
+
+```yaml
+source:
+  type: sqs
+
+
+  # Location
+  url: 
+  region: us-east-1
+  #endpoint: http://sqs:9324 useful for local testing
+
+  # Options
+  visibilityTimeoutInSeconds: 100 # defaults to 30
+```