Repository: argoproj-labs/old-argo-dataflow (language: python)

Issue:
HTTP sources should be TLS v1.2 by default
It might be easiest to generate a self-signed certificate in Golang, rather than bundle them as files (the way argo-server does), which cause various permission issues:

https://golang.org/src/crypto/tls/generate_cert.go


Discussion / hints:

Fix patch:
diff --git a/Dockerfile b/Dockerfile
index 27f788f6..33f33cca 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -28,6 +28,7 @@ ENTRYPOINT ["/manager"]
 
 FROM builder AS runner-builder
 ARG VERSION=unset
+RUN openssl req -x509 -newkey rsa:4096 -keyout runner.key -out runner.crt -days 365 -nodes -subj /CN=localhost/O=ArgoProj
 COPY kill/ kill/
 RUN CGO_ENABLED=0 go build -ldflags="-s -w" -o bin/kill ./kill
 COPY prestop/ prestop/
@@ -40,6 +41,8 @@ RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 go build -ldfl
 
 FROM gcr.io/distroless/static:nonroot AS runner
 WORKDIR /
+COPY --from=runner-builder --chown=9653:9653 /workspace/runner.crt .
+COPY --from=runner-builder --chown=9653:9653 /workspace/runner.key .
 COPY runtimes runtimes
 COPY --from=runner-builder /workspace/bin/kill /bin/kill
 COPY --from=runner-builder /workspace/bin/prestop /bin/prestop
diff --git a/api/v1alpha1/step_types.go b/api/v1alpha1/step_types.go
index bece4112..1aeabfa4 100644
--- a/api/v1alpha1/step_types.go
+++ b/api/v1alpha1/step_types.go
@@ -111,18 +111,19 @@ func (in Step) GetPodSpec(req GetPodSpecReq) corev1.PodSpec {
 				VolumeMounts:    volumeMounts,
 				Resources:       standardResources,
 				Ports: []corev1.ContainerPort{
-					{ContainerPort: 3569},
+					{ContainerPort: 3570},
 				},
 				ReadinessProbe: &corev1.Probe{
 					Handler: corev1.Handler{
-						HTTPGet: &corev1.HTTPGetAction{Path: "/ready", Port: intstr.FromInt(3569)},
+						HTTPGet: &corev1.HTTPGetAction{Scheme: "HTTPS", Path: "/ready", Port: intstr.FromInt(3570)},
 					},
 				},
 				Lifecycle: &corev1.Lifecycle{
 					PreStop: &corev1.Handler{
 						HTTPGet: &corev1.HTTPGetAction{
-							Path: "/pre-stop?source=kubernetes",
-							Port: intstr.FromInt(3569),
+							Path:   "/pre-stop?source=kubernetes",
+							Port:   intstr.FromInt(3570),
+							Scheme: "HTTPS",
 						},
 					},
 				},
diff --git a/docs/EXAMPLES.md b/docs/EXAMPLES.md
index 3fab4690..3ec20680 100644
--- a/docs/EXAMPLES.md
+++ b/docs/EXAMPLES.md
@@ -149,6 +149,19 @@ your code when the step starts. This example presents how one can use go runtime
 kubectl apply -f https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/106-git-go-pipeline.yaml
 ```
 
+### [106-git-nodejs](https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/106-git-nodejs-pipeline.yaml)
+
+This example of a pipeline using Git with NodeJS.
+
+The Git handler allows you to check your application source code into Git. Dataflow will checkout and build
+your code when the step starts. This example presents how one can use nodejs runtime git step.
+
+[Learn about Git steps](../docs/GIT.md)
+
+```
+kubectl apply -f https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/106-git-nodejs-pipeline.yaml
+```
+
 ### [106-git-python](https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/106-git-python-pipeline.yaml)
 
 This example of a pipeline using Git.
diff --git a/manager/controllers/step_controller.go b/manager/controllers/step_controller.go
index e3a6e9d7..e3e8d477 100644
--- a/manager/controllers/step_controller.go
+++ b/manager/controllers/step_controller.go
@@ -207,7 +207,7 @@ func (r *StepReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 				},
 				Spec: corev1.ServiceSpec{
 					Ports: []corev1.ServicePort{
-						{Port: 80, TargetPort: intstr.FromInt(3569)},
+						{Port: 443, TargetPort: intstr.FromInt(3570)},
 					},
 					Selector: map[string]string{
 						dfv1.KeyPipelineName: pipelineName,
diff --git a/runner/init/init.go b/runner/init/init.go
index 973f1cc9..1f78bb47 100644
--- a/runner/init/init.go
+++ b/runner/init/init.go
@@ -46,7 +46,7 @@ func Exec(ctx context.Context) error {
 	}
 	step := dfv1.Step{}
 	sharedutil.MustUnJSON(os.Getenv(dfv1.EnvStep), &step)
-	logger.Info("creating out authorization file")
+	logger.Info("creating authorization file")
 	if err := os.WriteFile(dfv1.PathAuthorization, []byte(sharedutil.RandString()), 0o600); sharedutil.IgnoreExist(err) != nil {
 		return fmt.Errorf("failed to create authorization file: %w", err)
 	}
diff --git a/runner/sidecar/in.go b/runner/sidecar/in.go
index afcb8bc7..25310710 100644
--- a/runner/sidecar/in.go
+++ b/runner/sidecar/in.go
@@ -76,7 +76,7 @@ func connectIn(ctx context.Context, sink func([]byte) error) (func(context.Conte
 				return fmt.Errorf("failed to send to main: %w", err)
 			} else {
 				body, _ := ioutil.ReadAll(resp.Body)
-				defer func() { _ = resp.Body.Close() }()
+				_ = resp.Body.Close()
 				if resp.StatusCode >= 300 {
 					return fmt.Errorf("failed to send to main: %q %q", resp.Status, body)
 				}
diff --git a/runner/sidecar/sidecar.go b/runner/sidecar/sidecar.go
index 9013275c..2604a4c0 100644
--- a/runner/sidecar/sidecar.go
+++ b/runner/sidecar/sidecar.go
@@ -2,6 +2,7 @@ package sidecar
 
 import (
 	"context"
+	"crypto/tls"
 	"net/http"
 	"os"
 	"strconv"
@@ -131,7 +132,7 @@ func Exec(ctx context.Context) error {
 
 	connectOut(toSinks)
 
-	server := &http.Server{Addr: ":3569"}
+	server := &http.Server{Addr: "localhost:3569"}
 	addStopHook(func(ctx context.Context) error {
 		logger.Info("closing HTTP server")
 		return server.Shutdown(context.Background())
@@ -139,11 +140,24 @@ func Exec(ctx context.Context) error {
 	go func() {
 		logger.Info("starting HTTP server")
 		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
-			logger.Error(err, "failed to listen-and-server")
+			logger.Error(err, "failed to listen-and-server on HTTP")
 		}
 		logger.Info("HTTP server shutdown")
 	}()
 
+	httpServer := &http.Server{Addr: ":3570", TLSConfig: &tls.Config{MinVersion: tls.VersionTLS12}}
+	addStopHook(func(ctx context.Context) error {
+		logger.Info("closing HTTPS server")
+		return httpServer.Shutdown(context.Background())
+	})
+	go func() {
+		logger.Info("starting HTTPS server")
+		if err := httpServer.ListenAndServeTLS("runner.crt", "runner.key"); err != nil && err != http.ErrServerClosed {
+			logger.Error(err, "failed to listen-and-server on HTTPS")
+		}
+		logger.Info("HTTPS server shutdown")
+	}()
+
 	toMain, err := connectIn(ctx, toSinks)
 	if err != nil {
 		return err
diff --git a/runner/sidecar/source/s3/s3.go b/runner/sidecar/source/s3/s3.go
index 46918697..688d7075 100644
--- a/runner/sidecar/source/s3/s3.go
+++ b/runner/sidecar/source/s3/s3.go
@@ -3,6 +3,7 @@ package s3
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"fmt"
 	dfv1 "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
 	"github.com/argoproj-labs/argo-dataflow/runner/sidecar/source"
@@ -76,8 +77,14 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, pipelineNa
 	jobs := workqueue.New()
 	authorization := sharedutil.RandString()
 	if leadReplica {
-		endpoint := "http://" + pipelineName + "-" + stepName + "/sources/" + sourceName
-		logger.Info("starting lead workers", "source", sourceName)
+		endpoint := "https://" + pipelineName + "-" + stepName + "/sources/" + sourceName
+		logger.Info("starting lead workers", "source", sourceName, "endpoint", endpoint)
+		t := http.DefaultTransport.(*http.Transport).Clone()
+		t.MaxIdleConns = 100
+		t.MaxConnsPerHost = 100
+		t.MaxIdleConnsPerHost = 100
+		t.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
+		httpClient := &http.Client{Timeout: 10 * time.Second, Transport: t}
 		// create N workers to support concurrency
 		for w := 0; w < concurrency; w++ {
 			go func() {
@@ -95,7 +102,7 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, pipelineNa
 							panic(err)
 						}
 						req.Header.Set("Authorization", authorization)
-						resp, err := http.DefaultClient.Do(req)
+						resp, err := httpClient.Do(req)
 						if err != nil {
 							logger.Error(err, "failed to process object", "key", key)
 						} else {


Test patch:
diff --git a/api/v1alpha1/step_types_test.go b/api/v1alpha1/step_types_test.go
index a06c0130..ac7ee455 100644
--- a/api/v1alpha1/step_types_test.go
+++ b/api/v1alpha1/step_types_test.go
@@ -62,14 +62,15 @@ func TestStep_GetPodSpec(t *testing.T) {
 						Name:            "sidecar",
 						Lifecycle: &corev1.Lifecycle{PreStop: &corev1.Handler{
 							HTTPGet: &corev1.HTTPGetAction{
-								Path: "/pre-stop?source=kubernetes",
-								Port: intstr.FromInt(3569),
+								Path:   "/pre-stop?source=kubernetes",
+								Port:   intstr.FromInt(3570),
+								Scheme: "HTTPS",
 							},
 						}},
-						Ports: []corev1.ContainerPort{{ContainerPort: 3569}},
+						Ports: []corev1.ContainerPort{{ContainerPort: 3570}},
 						ReadinessProbe: &corev1.Probe{
 							Handler: corev1.Handler{
-								HTTPGet: &corev1.HTTPGetAction{Path: "/ready", Port: intstr.FromInt(3569)},
+								HTTPGet: &corev1.HTTPGetAction{Path: "/ready", Port: intstr.FromInt(3570), Scheme: "HTTPS"},
 							},
 						},
 						Resources:       standardResources,
diff --git a/test/http.go b/test/http.go
index 1195692b..562860aa 100644
--- a/test/http.go
+++ b/test/http.go
@@ -37,7 +37,7 @@ func PumpHTTP(_url, prefix string, n int, sleep time.Duration) {
 func PumpHTTPTolerantly(n int) {
 	for i := 0; i < n; {
 		CatchPanic(func() {
-			PumpHTTP("http://http-main/sources/default", fmt.Sprintf("my-msg-%d", i), 1, 0)
+			PumpHTTP("https://http-main/sources/default", fmt.Sprintf("my-msg-%d", i), 1, 0)
 			i++
 		}, func(err error) {
 			log.Printf("ignoring: %v\n", err)
diff --git a/test/service.go b/test/service.go
index 861a4a7d..1b716f16 100644
--- a/test/service.go
+++ b/test/service.go
@@ -26,7 +26,7 @@ func WaitForService() {
 	for _, x := range list.Items {
 		if _, ok := x.Spec.Selector[KeyPipelineName]; ok {
 			log.Printf("waiting for service %q\n", x.Name)
-			InvokeTestAPI("/http/wait-for?url=%s", url.QueryEscape("http://"+x.Name))
+			InvokeTestAPI("/http/wait-for?url=%s", url.QueryEscape("https://"+x.Name))
 		}
 	}
 }
diff --git a/test/stress/http_stress_test.go b/test/stress/http_stress_test.go
index da0d362b..726adcd0 100644
--- a/test/stress/http_stress_test.go
+++ b/test/stress/http_stress_test.go
@@ -40,7 +40,7 @@ func TestHTTPSourceStress(t *testing.T) {
 	defer StartMetricsLogger()()
 	defer StartTPSReporter(t, "main", prefix, n)()
 
-	go PumpHTTP("http://http-main/sources/default", prefix, n, 0)
+	go PumpHTTP("https://http-main/sources/default", prefix, n, 0)
 	WaitForStep(TotalSunkMessages(n), params.timeout)
 
 }
@@ -73,6 +73,6 @@ func TestHTTPSinkStress(t *testing.T) {
 	defer StartMetricsLogger()()
 	defer StartTPSReporter(t, "main", prefix, n)()
 
-	go PumpHTTP("http://http-main/sources/default", prefix, n, 0)
+	go PumpHTTP("https://http-main/sources/default", prefix, n, 0)
 	WaitForStep(TotalSunkMessages(n*2), params.timeout)
 }
diff --git a/test/stress/test-results.json b/test/stress/test-results.json
index 138ca257..dd29221e 100644
--- a/test/stress/test-results.json
+++ b/test/stress/test-results.json
@@ -3,7 +3,7 @@
   "TestHTTPSinkStress/currentContext=docker-desktop,replicas=2,n=10000.tps": 400,
   "TestHTTPSinkStress/currentContext=gke_jesse-sb_us-west2_cluster-2,replicas=1,n=10000.tps": 950,
   "TestHTTPSinkStress/currentContext=gke_jesse-sb_us-west2_cluster-2,replicas=2,n=10000.tps": 650,
-  "TestHTTPSourceStress/currentContext=docker-desktop,replicas=1,n=10000.tps": 400,
+  "TestHTTPSourceStress/currentContext=docker-desktop,replicas=1,n=10000.tps": 300,
   "TestHTTPSourceStress/currentContext=docker-desktop,replicas=2,n=10000.tps": 850,
   "TestHTTPSourceStress/currentContext=gke_jesse-sb_us-west2_cluster-2,replicas=1,n=10000.tps": 1650,
   "TestHTTPSourceStress/currentContext=gke_jesse-sb_us-west2_cluster-2,replicas=2,n=10000.tps": 2300,
diff --git a/testapi/http.go b/testapi/http.go
index 840995ae..06d2a67c 100644
--- a/testapi/http.go
+++ b/testapi/http.go
@@ -2,7 +2,9 @@ package main
 
 import (
 	"bytes"
+	"crypto/tls"
 	"fmt"
+	"io"
 	"net/http"
 	"strconv"
 	"time"
@@ -14,6 +16,12 @@ type req struct {
 }
 
 func init() {
+	t := http.DefaultTransport.(*http.Transport).Clone()
+	t.MaxIdleConns = 100
+	t.MaxConnsPerHost = 100
+	t.MaxIdleConnsPerHost = 100
+	t.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
+	httpClient := &http.Client{Timeout: 10 * time.Second, Transport: t}
 	http.HandleFunc("/http/pump", func(w http.ResponseWriter, r *http.Request) {
 		url := r.URL.Query().Get("url")
 		duration, err := time.ParseDuration(r.URL.Query().Get("sleep"))
@@ -43,12 +51,16 @@ func init() {
 					results <- err
 				} else {
 					req.Header.Set("Authorization", "Bearer my-bearer-token")
-					if resp, err := http.DefaultClient.Do(req); err != nil {
+					if resp, err := httpClient.Do(req); err != nil {
 						results <- err
-					} else if resp.StatusCode >= 300 {
-						results <- fmt.Errorf("%s", resp.Status)
 					} else {
-						results <- fmt.Sprintf("sent %q (%d/%d, %.0f TPS) to %q", j.msg, j.i+1, n, (1+float64(j.i))/time.Since(start).Seconds(), url)
+						_, _ = io.Copy(io.Discard, resp.Body) // needed for keep alive
+						_ = resp.Body.Close()
+						if resp.StatusCode >= 300 {
+							results <- fmt.Errorf("%s", resp.Status)
+						} else {
+							results <- fmt.Sprintf("sent %q (%d/%d, %.0f TPS) to %q", j.msg, j.i+1, n, (1+float64(j.i))/time.Since(start).Seconds(), url)
+						}
 					}
 				}
 			}
@@ -79,8 +91,7 @@ func init() {
 		url := r.URL.Query().Get("url")
 		w.WriteHeader(200)
 		for {
-			_, err := http.Get(url)
-			if err != nil {
+			if _, err := httpClient.Get(url); err != nil {
 				_, _ = fmt.Fprintf(w, "%q is not ready: %v\n", url, err)
 			} else {
 				return