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

Issue:
Add authentication to S3/HTTP endpoints
Currently these endpoints do not have any authentication. S3 should require a bearer token, which can be randomly generated, HTTP should be specified in a secret.


Discussion / hints:

Fix patch:
diff --git a/Makefile b/Makefile
index 445e8b62..b6da316d 100644
--- a/Makefile
+++ b/Makefile
@@ -66,7 +66,7 @@ $(GOBIN)/goreman:
 	go install github.com/mattn/goreman@v0.3.7
 
 # Run against the configured Kubernetes cluster in ~/.kube/config
-start: build deploy $(GOBIN)/goreman
+start: build runner deploy $(GOBIN)/goreman
 	kubectl config set-context --current --namespace=argo-dataflow-system
 	goreman -set-ports=false -logtime=false start
 wait:
diff --git a/docs/EXAMPLES.md b/docs/EXAMPLES.md
index 75418039..3fab4690 100644
--- a/docs/EXAMPLES.md
+++ b/docs/EXAMPLES.md
@@ -348,6 +348,15 @@ This example has two sources
 kubectl apply -f https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/301-two-sources-pipeline.yaml
 ```
 
+### [dataflow-http-main-source-default](https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/dataflow-103-http-main-source-default-secret.yaml)
+
+This is an example of providing a bearer token for a HTTP source.
+
+
+```
+kubectl apply -f https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/dataflow-103-http-main-source-default-secret.yaml
+```
+
 ### [dataflow-kafka-default](https://raw.githubusercontent.com/argoproj-labs/argo-dataflow/main/examples/dataflow-kafka-default-secret.yaml)
 
 This is an example of providing a namespace named Kafka configuration.
diff --git a/examples/301-http-pipeline.py b/examples/301-http-pipeline.py
index 8a4a4b78..df32dd5e 100644
--- a/examples/301-http-pipeline.py
+++ b/examples/301-http-pipeline.py
@@ -12,6 +12,7 @@
 * Adding replicas will nearly linearly increase throughput.
 * Due to the lack of state, do not use HTTP sources and sinks to connect steps. 
 """)
+     .annotate("dataflow.argoproj.io/needs", "dataflow-103-http-main-source-default-secret.yaml")
      .step(
         (cron('*/3 * * * * *')
          .cat('cron')
diff --git a/examples/301-http-pipeline.yaml b/examples/301-http-pipeline.yaml
index 657df862..5066ee14 100644
--- a/examples/301-http-pipeline.yaml
+++ b/examples/301-http-pipeline.yaml
@@ -8,6 +8,7 @@ metadata:
       \ may return and error code, clients will need to re-send messages.\n* Adding\
       \ replicas will nearly linearly increase throughput.\n* Due to the lack of state,\
       \ do not use HTTP sources and sinks to connect steps. \n"
+    dataflow.argoproj.io/needs: dataflow-103-http-main-source-default-secret.yaml
     dataflow.argoproj.io/owner: argoproj-labs
   name: 301-http
 spec:
diff --git a/examples/dataflow-103-http-main-source-default-secret.yaml b/examples/dataflow-103-http-main-source-default-secret.yaml
new file mode 100644
index 00000000..b99c0f90
--- /dev/null
+++ b/examples/dataflow-103-http-main-source-default-secret.yaml
@@ -0,0 +1,9 @@
+kind: Secret
+apiVersion: v1
+metadata:
+  name: dataflow-http-main-source-default
+  annotations:
+    dataflow.argoproj.io/description: |
+      This is an example of providing a bearer token for a HTTP source.
+stringData:
+  authorization: Bearer my-bearer-token
\ No newline at end of file
diff --git a/runner/sidecar/kafka.go b/runner/sidecar/kafka.go
index 69008861..9b1ac59a 100644
--- a/runner/sidecar/kafka.go
+++ b/runner/sidecar/kafka.go
@@ -8,7 +8,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
 func kafkaFromSecret(k *dfv1.Kafka, secret *corev1.Secret) error {
@@ -100,8 +99,8 @@ func saslFromSecret(secret *corev1.Secret) *dfv1.SASL {
 	return sasl
 }
 
-func enrichKafka(ctx context.Context, secrets v1.SecretInterface, x *dfv1.Kafka) error {
-	secret, err := secrets.Get(ctx, "dataflow-kafka-"+x.Name, metav1.GetOptions{})
+func enrichKafka(ctx context.Context, x *dfv1.Kafka) error {
+	secret, err := secretInterface.Get(ctx, "dataflow-kafka-"+x.Name, metav1.GetOptions{})
 	if err != nil {
 		if !apierr.IsNotFound(err) {
 			return err
diff --git a/runner/sidecar/s3.go b/runner/sidecar/s3.go
index fe3dbd1e..bb1fa305 100644
--- a/runner/sidecar/s3.go
+++ b/runner/sidecar/s3.go
@@ -6,7 +6,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
 func s3FromSecret(x *dfv1.S3, secret *corev1.Secret) error {
@@ -35,8 +34,8 @@ func s3FromSecret(x *dfv1.S3, secret *corev1.Secret) error {
 	return nil
 }
 
-func enrichS3(ctx context.Context, secrets v1.SecretInterface, x *dfv1.S3) error {
-	secret, err := secrets.Get(ctx, "dataflow-s3-"+x.Name, metav1.GetOptions{})
+func enrichS3(ctx context.Context, x *dfv1.S3) error {
+	secret, err := secretInterface.Get(ctx, "dataflow-s3-"+x.Name, metav1.GetOptions{})
 	if err != nil {
 		if !apierr.IsNotFound(err) {
 			return err
diff --git a/runner/sidecar/shared/kafka/kafka.go b/runner/sidecar/shared/kafka/kafka.go
index 54dd5264..597735f3 100644
--- a/runner/sidecar/shared/kafka/kafka.go
+++ b/runner/sidecar/shared/kafka/kafka.go
@@ -5,14 +5,14 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 
 	"github.com/Shopify/sarama"
 	dfv1 "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 )
 
-func NewConfig(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, k dfv1.Kafka) (*sarama.Config, error) {
+func NewConfig(ctx context.Context, secretInterface corev1.SecretInterface, k dfv1.Kafka) (*sarama.Config, error) {
 	x := sarama.NewConfig()
 	x.ClientID = dfv1.CtrSidecar
 	if k.Version != "" {
@@ -24,7 +24,7 @@ func NewConfig(ctx context.Context, kubernetesInterface kubernetes.Interface, na
 	}
 	if k.NET != nil {
 		if k.NET.TLS != nil {
-			tlsConfig, err := getTLSConfig(ctx, kubernetesInterface, namespace, k)
+			tlsConfig, err := getTLSConfig(ctx, secretInterface, k)
 			if err != nil {
 				return nil, fmt.Errorf("failed to load tls config, %w", err)
 			}
@@ -36,8 +36,7 @@ func NewConfig(ctx context.Context, kubernetesInterface kubernetes.Interface, na
 			if sasl.UserSecret == nil || sasl.PasswordSecret == nil {
 				return nil, fmt.Errorf("invalid sasl config, user secret or password secret not configured")
 			}
-			secrets := kubernetesInterface.CoreV1().Secrets(namespace)
-			if userSecret, err := secrets.Get(ctx, sasl.UserSecret.Name, metav1.GetOptions{}); err != nil {
+			if userSecret, err := secretInterface.Get(ctx, sasl.UserSecret.Name, metav1.GetOptions{}); err != nil {
 				return nil, fmt.Errorf("failed to get user secret, %w", err)
 			} else {
 				if d, ok := userSecret.Data[sasl.UserSecret.Key]; !ok {
@@ -46,7 +45,7 @@ func NewConfig(ctx context.Context, kubernetesInterface kubernetes.Interface, na
 					x.Net.SASL.User = string(d)
 				}
 			}
-			if passwordSecret, err := secrets.Get(ctx, sasl.PasswordSecret.Name, metav1.GetOptions{}); err != nil {
+			if passwordSecret, err := secretInterface.Get(ctx, sasl.PasswordSecret.Name, metav1.GetOptions{}); err != nil {
 				return nil, fmt.Errorf("failed to get password secret, %w", err)
 			} else {
 				if d, ok := passwordSecret.Data[sasl.PasswordSecret.Key]; !ok {
@@ -62,16 +61,15 @@ func NewConfig(ctx context.Context, kubernetesInterface kubernetes.Interface, na
 	return x, nil
 }
 
-func getTLSConfig(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, k dfv1.Kafka) (*tls.Config, error) {
+func getTLSConfig(ctx context.Context, secretInterface corev1.SecretInterface, k dfv1.Kafka) (*tls.Config, error) {
 	t := k.NET.TLS
 	if t == nil {
 		return nil, fmt.Errorf("tls config not found")
 	}
 
 	c := &tls.Config{}
-	secrets := kubernetesInterface.CoreV1().Secrets(namespace)
 	if t.CACertSecret != nil {
-		if caCertSecret, err := secrets.Get(ctx, t.CACertSecret.Name, metav1.GetOptions{}); err != nil {
+		if caCertSecret, err := secretInterface.Get(ctx, t.CACertSecret.Name, metav1.GetOptions{}); err != nil {
 			return nil, fmt.Errorf("failed to get ca cert secret, %w", err)
 		} else {
 			if d, ok := caCertSecret.Data[t.CACertSecret.Key]; !ok {
@@ -85,7 +83,7 @@ func getTLSConfig(ctx context.Context, kubernetesInterface kubernetes.Interface,
 	}
 	if t.CertSecret != nil && t.KeySecret != nil {
 		var certBytes, keyBytes []byte
-		if certSecret, err := secrets.Get(ctx, t.CertSecret.Name, metav1.GetOptions{}); err != nil {
+		if certSecret, err := secretInterface.Get(ctx, t.CertSecret.Name, metav1.GetOptions{}); err != nil {
 			return nil, fmt.Errorf("failed to get cert secret, %w", err)
 		} else {
 			if d, ok := certSecret.Data[t.CertSecret.Key]; !ok {
@@ -94,7 +92,7 @@ func getTLSConfig(ctx context.Context, kubernetesInterface kubernetes.Interface,
 				certBytes = d
 			}
 		}
-		if keySecret, err := secrets.Get(ctx, t.KeySecret.Name, metav1.GetOptions{}); err != nil {
+		if keySecret, err := secretInterface.Get(ctx, t.KeySecret.Name, metav1.GetOptions{}); err != nil {
 			return nil, fmt.Errorf("failed to get key secret, %w", err)
 		} else {
 			if d, ok := keySecret.Data[t.KeySecret.Key]; !ok {
diff --git a/runner/sidecar/shared/stan/stan_conn.go b/runner/sidecar/shared/stan/stan_conn.go
index 50a0ce8a..bf9dff3b 100644
--- a/runner/sidecar/shared/stan/stan_conn.go
+++ b/runner/sidecar/shared/stan/stan_conn.go
@@ -3,13 +3,12 @@ package stan
 import (
 	"context"
 	"fmt"
-
 	dfv1 "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
 	"github.com/argoproj-labs/argo-dataflow/shared/util"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/stan.go"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
 var logger = util.NewLogger()
@@ -50,7 +49,7 @@ func (nsc *Conn) QueueSubscribe(subject, qgroup string, cb stan.MsgHandler, opts
 	return nsc.sc.QueueSubscribe(subject, qgroup, cb, opts...)
 }
 
-func ConnectSTAN(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, x dfv1.STAN, clientID string) (*Conn, error) {
+func ConnectSTAN(ctx context.Context, secretInterface corev1.SecretInterface, x dfv1.STAN, clientID string) (*Conn, error) {
 	conn := &Conn{}
 	opts := []nats.Option{
 		// Do not reconnect here but handle reconnction outside
@@ -66,7 +65,7 @@ func ConnectSTAN(ctx context.Context, kubernetesInterface kubernetes.Interface,
 	}
 	switch x.AuthStrategy() {
 	case dfv1.STANAuthToken:
-		token, err := getSTANAuthToken(ctx, kubernetesInterface, namespace, x)
+		token, err := getSTANAuthToken(ctx, secretInterface, x)
 		if err != nil {
 			return nil, err
 		}
@@ -96,15 +95,14 @@ func ConnectSTAN(ctx context.Context, kubernetesInterface kubernetes.Interface,
 	return conn, nil
 }
 
-func getSTANAuthToken(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, x dfv1.STAN) (string, error) {
+func getSTANAuthToken(ctx context.Context, secretInterface corev1.SecretInterface, x dfv1.STAN) (string, error) {
 	if x.AuthStrategy() != dfv1.STANAuthToken {
 		return "", fmt.Errorf("auth strategy is not token but %s", x.AuthStrategy())
 	}
 	if x.Auth == nil || x.Auth.Token == nil {
 		return "", fmt.Errorf("token secret selector is nil")
 	}
-	secrets := kubernetesInterface.CoreV1().Secrets(namespace)
-	secret, err := secrets.Get(ctx, x.Auth.Token.Name, metav1.GetOptions{})
+	secret, err := secretInterface.Get(ctx, x.Auth.Token.Name, metav1.GetOptions{})
 	if err != nil {
 		return "", err
 	}
diff --git a/runner/sidecar/sidecar.go b/runner/sidecar/sidecar.go
index a57761f3..9013275c 100644
--- a/runner/sidecar/sidecar.go
+++ b/runner/sidecar/sidecar.go
@@ -20,7 +20,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
-	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	ctrl "sigs.k8s.io/controller-runtime"
 )
 
@@ -33,6 +33,7 @@ var (
 	dynamicInterface    dynamic.Interface
 	lastStep            dfv1.Step
 	kubernetesInterface kubernetes.Interface
+	secretInterface     corev1.SecretInterface
 	prePatchHooks       []func(ctx context.Context) error // hooks to run before patching
 	replica             int
 	step                dfv1.Step // this is updated on start, and then periodically as we update the status
@@ -49,6 +50,7 @@ func Exec(ctx context.Context) error {
 	restConfig := ctrl.GetConfigOrDie()
 	dynamicInterface = dynamic.NewForConfigOrDie(restConfig)
 	kubernetesInterface = kubernetes.NewForConfigOrDie(restConfig)
+	secretInterface = kubernetesInterface.CoreV1().Secrets(namespace)
 
 	sharedutil.MustUnJSON(os.Getenv(dfv1.EnvStep), &step)
 
@@ -233,29 +235,26 @@ func patchStepStatus() {
 }
 
 func enrichSpec(ctx context.Context) error {
-	secrets := kubernetesInterface.CoreV1().Secrets(namespace)
-
-	if err := enrichSources(ctx, secrets); err != nil {
+	if err := enrichSources(ctx); err != nil {
 		return err
 	}
-
-	return enrichSinks(ctx, secrets)
+	return enrichSinks(ctx)
 }
 
-func enrichSources(ctx context.Context, secrets v1.SecretInterface) error {
+func enrichSources(ctx context.Context) error {
 	for i, source := range step.Spec.Sources {
 		if x := source.STAN; x != nil {
-			if err := enrichSTAN(ctx, secrets, x); err != nil {
+			if err := enrichSTAN(ctx, x); err != nil {
 				return err
 			}
 			source.STAN = x
 		} else if x := source.Kafka; x != nil {
-			if err := enrichKafka(ctx, secrets, &x.Kafka); err != nil {
+			if err := enrichKafka(ctx, &x.Kafka); err != nil {
 				return err
 			}
 			source.Kafka = x
 		} else if x := source.S3; x != nil {
-			if err := enrichS3(ctx, secrets, &x.S3); err != nil {
+			if err := enrichS3(ctx, &x.S3); err != nil {
 				return err
 			}
 			source.S3 = x
@@ -265,20 +264,20 @@ func enrichSources(ctx context.Context, secrets v1.SecretInterface) error {
 	return nil
 }
 
-func enrichSinks(ctx context.Context, secrets v1.SecretInterface) error {
+func enrichSinks(ctx context.Context) error {
 	for i, sink := range step.Spec.Sinks {
 		if x := sink.STAN; x != nil {
-			if err := enrichSTAN(ctx, secrets, x); err != nil {
+			if err := enrichSTAN(ctx, x); err != nil {
 				return err
 			}
 			sink.STAN = x
 		} else if x := sink.Kafka; x != nil {
-			if err := enrichKafka(ctx, secrets, x); err != nil {
+			if err := enrichKafka(ctx, x); err != nil {
 				return err
 			}
 			sink.Kafka = x
 		} else if x := sink.S3; x != nil {
-			if err := enrichS3(ctx, secrets, &x.S3); err != nil {
+			if err := enrichS3(ctx, &x.S3); err != nil {
 				return err
 			}
 			sink.S3 = x
diff --git a/runner/sidecar/sink/db/db.go b/runner/sidecar/sink/db/db.go
index 2eddc341..7a15eebb 100644
--- a/runner/sidecar/sink/db/db.go
+++ b/runner/sidecar/sink/db/db.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"time"
 
 	"github.com/antonmedv/expr"
@@ -14,7 +15,6 @@ import (
 	sharedutil "github.com/argoproj-labs/argo-dataflow/shared/util"
 	_ "github.com/go-sql-driver/mysql"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 )
 
 var logger = sharedutil.NewLogger()
@@ -25,8 +25,8 @@ type dbSink struct {
 	progs   map[string]*vm.Program
 }
 
-func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, x dfv1.DBSink) (sink.Interface, error) {
-	dataSource, err := getDataSource(ctx, kubernetesInterface, namespace, x)
+func New(ctx context.Context, secretInterface corev1.SecretInterface, x dfv1.DBSink) (sink.Interface, error) {
+	dataSource, err := getDataSource(ctx, secretInterface, x)
 	if err != nil {
 		return nil, fmt.Errorf("failed to find data source: %w", err)
 	}
@@ -122,12 +122,11 @@ func (d dbSink) execStatement(tx *sql.Tx, sql string, args []string, msg []byte)
 	return rs, nil
 }
 
-func getDataSource(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, x dfv1.DBSink) (string, error) {
+func getDataSource(ctx context.Context, secretInterface corev1.SecretInterface, x dfv1.DBSink) (string, error) {
 	if x.DataSource.Value != "" {
 		return x.DataSource.Value, nil
 	}
 	if x.DataSource.ValueFrom != nil && x.DataSource.ValueFrom.SecretKeyRef != nil {
-		secretInterface := kubernetesInterface.CoreV1().Secrets(namespace)
 		secret, err := secretInterface.Get(ctx, x.DataSource.ValueFrom.SecretKeyRef.Name, metav1.GetOptions{})
 		if err != nil {
 			return "", fmt.Errorf("failed to get secret %q: %w", x.DataSource.ValueFrom.SecretKeyRef.Name, err)
diff --git a/runner/sidecar/sink/http/http.go b/runner/sidecar/sink/http/http.go
index a00e896d..65b36d0d 100644
--- a/runner/sidecar/sink/http/http.go
+++ b/runner/sidecar/sink/http/http.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"net/http"
 	"time"
 
@@ -12,7 +13,6 @@ import (
 	"github.com/argoproj-labs/argo-dataflow/runner/sidecar/sink"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
 )
 
 type httpSink struct {
@@ -21,14 +21,14 @@ type httpSink struct {
 	url    string
 }
 
-func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, x dfv1.HTTPSink) (sink.Interface, error) {
+func New(ctx context.Context, secretInterface corev1.SecretInterface, x dfv1.HTTPSink) (sink.Interface, error) {
 	header := http.Header{}
 	for _, h := range x.Headers {
 		if h.Value != "" {
 			header.Add(h.Name, h.Value)
 		} else if h.ValueFrom != nil {
 			r := h.ValueFrom.SecretKeyRef
-			secret, err := kubernetesInterface.CoreV1().Secrets(namespace).Get(ctx, r.Name, metav1.GetOptions{})
+			secret, err := secretInterface.Get(ctx, r.Name, metav1.GetOptions{})
 			if err != nil {
 				return nil, fmt.Errorf("failed to get secret %q: %w", r.Name, err)
 			}
diff --git a/runner/sidecar/sink/kafka/kafka.go b/runner/sidecar/sink/kafka/kafka.go
index a166d75f..914584b4 100644
--- a/runner/sidecar/sink/kafka/kafka.go
+++ b/runner/sidecar/sink/kafka/kafka.go
@@ -2,12 +2,12 @@ package kafka
 
 import (
 	"context"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 
 	"github.com/Shopify/sarama"
 	dfv1 "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
 	"github.com/argoproj-labs/argo-dataflow/runner/sidecar/shared/kafka"
 	"github.com/argoproj-labs/argo-dataflow/runner/sidecar/sink"
-	"k8s.io/client-go/kubernetes"
 )
 
 type kafkaSink struct {
@@ -15,8 +15,8 @@ type kafkaSink struct {
 	topic    string
 }
 
-func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, x dfv1.Kafka) (sink.Interface, error) {
-	config, err := kafka.NewConfig(ctx, kubernetesInterface, namespace, x)
+func New(ctx context.Context, secretInterface corev1.SecretInterface, x dfv1.Kafka) (sink.Interface, error) {
+	config, err := kafka.NewConfig(ctx, secretInterface, x)
 	if err != nil {
 		return nil, err
 	}
diff --git a/runner/sidecar/sink/s3/s3.go b/runner/sidecar/sink/s3/s3.go
index c75abbc9..06b0344e 100644
--- a/runner/sidecar/sink/s3/s3.go
+++ b/runner/sidecar/sink/s3/s3.go
@@ -10,9 +10,8 @@ import (
 	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"os"
-
-	"k8s.io/client-go/kubernetes"
 )
 
 type s3Sink struct {
@@ -25,8 +24,7 @@ type message struct {
 	Path string `json:"path"`
 }
 
-func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, x dfv1.S3Sink) (sink.Interface, error) {
-	secretInterface := kubernetesInterface.CoreV1().Secrets(namespace)
+func New(ctx context.Context, secretInterface v1.SecretInterface, x dfv1.S3Sink) (sink.Interface, error) {
 	var accessKeyID string
 	{
 		secretName := x.Credentials.AccessKeyID.Name
diff --git a/runner/sidecar/sink/stan/stan.go b/runner/sidecar/sink/stan/stan.go
index e78f61de..d23f7373 100644
--- a/runner/sidecar/sink/stan/stan.go
+++ b/runner/sidecar/sink/stan/stan.go
@@ -3,6 +3,7 @@ package stan
 import (
 	"context"
 	"fmt"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"math/rand"
 	"time"
 
@@ -11,7 +12,6 @@ import (
 	"github.com/argoproj-labs/argo-dataflow/runner/sidecar/sink"
 	sharedutil "github.com/argoproj-labs/argo-dataflow/shared/util"
 	runtimeutil "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/client-go/kubernetes"
 )
 
 var logger = sharedutil.NewLogger()
@@ -21,7 +21,7 @@ type stanSink struct {
 	subject string
 }
 
-func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace string, pipelineName, stepName string, replica int, sinkName string, x dfv1.STAN) (sink.Interface, error) {
+func New(ctx context.Context, secretInterface corev1.SecretInterface, pipelineName, stepName string, replica int, sinkName string, x dfv1.STAN) (sink.Interface, error) {
 	genClientID := func() string {
 		// In a particular situation, the stan connection status is inconsistent between stan server and client,
 		// the connection is lost from client side, but the server still thinks it's alive. In this case, use
@@ -34,7 +34,7 @@ func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespac
 	var conn *stan.Conn
 	var err error
 	clientID := genClientID()
-	conn, err = stan.ConnectSTAN(ctx, kubernetesInterface, namespace, x, clientID)
+	conn, err = stan.ConnectSTAN(ctx, secretInterface, x, clientID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to stan url=%s clusterID=%s clientID=%s subject=%s: %w", x.NATSURL, x.ClusterID, clientID, x.Subject, err)
 	}
@@ -52,7 +52,7 @@ func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespac
 				if conn == nil || conn.IsClosed() {
 					logger.Info("stan connection lost, reconnecting...", "sink", sinkName)
 					clientID := genClientID()
-					conn, err = stan.ConnectSTAN(ctx, kubernetesInterface, namespace, x, clientID)
+					conn, err = stan.ConnectSTAN(ctx, secretInterface, x, clientID)
 					if err != nil {
 						logger.Error(err, "failed to reconnect", "sink", sinkName, "clientID", clientID)
 						continue
diff --git a/runner/sidecar/sinks.go b/runner/sidecar/sinks.go
index 10fedeff..45ca4e09 100644
--- a/runner/sidecar/sinks.go
+++ b/runner/sidecar/sinks.go
@@ -28,13 +28,13 @@ func connectSinks(ctx context.Context) (func([]byte) error, error) {
 		}
 		rateCounters[sinkName] = ratecounter.NewRateCounter(updateInterval)
 		if x := sink.STAN; x != nil {
-			if y, err := stan.New(ctx, kubernetesInterface, namespace, pipelineName, stepName, replica, sinkName, *x); err != nil {
+			if y, err := stan.New(ctx, secretInterface, pipelineName, stepName, replica, sinkName, *x); err != nil {
 				return nil, err
 			} else {
 				sinks[sinkName] = y
 			}
 		} else if x := sink.Kafka; x != nil {
-			if y, err := kafka.New(ctx, kubernetesInterface, namespace, *x); err != nil {
+			if y, err := kafka.New(ctx, secretInterface, *x); err != nil {
 				return nil, err
 			} else {
 				sinks[sinkName] = y
@@ -42,19 +42,19 @@ func connectSinks(ctx context.Context) (func([]byte) error, error) {
 		} else if x := sink.Log; x != nil {
 			sinks[sinkName] = logsink.New()
 		} else if x := sink.HTTP; x != nil {
-			if y, err := http.New(ctx, kubernetesInterface, namespace, *x); err != nil {
+			if y, err := http.New(ctx, secretInterface, *x); err != nil {
 				return nil, err
 			} else {
 				sinks[sinkName] = y
 			}
 		} else if x := sink.S3; x != nil {
-			if y, err := s3sink.New(ctx, kubernetesInterface, namespace, *x); err != nil {
+			if y, err := s3sink.New(ctx, secretInterface, *x); err != nil {
 				return nil, err
 			} else {
 				sinks[sinkName] = y
 			}
 		} else if x := sink.DB; x != nil {
-			if y, err := dbsink.New(ctx, kubernetesInterface, namespace, *x); err != nil {
+			if y, err := dbsink.New(ctx, secretInterface, *x); err != nil {
 				return nil, err
 			} else {
 				sinks[sinkName] = y
diff --git a/runner/sidecar/source/http/http.go b/runner/sidecar/source/http/http.go
index f703da73..91264774 100644
--- a/runner/sidecar/source/http/http.go
+++ b/runner/sidecar/source/http/http.go
@@ -12,10 +12,13 @@ type httpSource struct {
 	ready bool
 }
 
-func New(sourceName string, f source.Func) source.Interface {
+func New(sourceName, authorization string, f source.Func) source.Interface {
 	h := &httpSource{ready: true}
 	http.HandleFunc("/sources/"+sourceName, func(w http.ResponseWriter, r *http.Request) {
-		// TODO -auth
+		if r.Header.Get("Authorization") != authorization {
+			w.WriteHeader(403)
+			return
+		}
 		if !h.ready { // if we are not ready, we cannot serve requests
 			w.WriteHeader(503)
 			_, _ = w.Write([]byte("not ready"))
diff --git a/runner/sidecar/source/kafka/kafka.go b/runner/sidecar/source/kafka/kafka.go
index 83c2e05c..8f769802 100644
--- a/runner/sidecar/source/kafka/kafka.go
+++ b/runner/sidecar/source/kafka/kafka.go
@@ -3,6 +3,7 @@ package kafka
 import (
 	"context"
 	"fmt"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"time"
 
 	"github.com/argoproj-labs/argo-dataflow/runner/sidecar/shared/kafka"
@@ -12,7 +13,6 @@ import (
 	"github.com/argoproj-labs/argo-dataflow/runner/sidecar/source"
 	sharedutil "github.com/argoproj-labs/argo-dataflow/shared/util"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"k8s.io/client-go/kubernetes"
 )
 
 var logger = sharedutil.NewLogger()
@@ -45,9 +45,9 @@ func (h handler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.Con
 	return nil
 }
 
-func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace, pipelineName, stepName, sourceName string, x dfv1.KafkaSource, f source.Func) (source.Interface, error) {
+func New(ctx context.Context, secretInterface corev1.SecretInterface, pipelineName, stepName, sourceName string, x dfv1.KafkaSource, f source.Func) (source.Interface, error) {
 	groupName := pipelineName + "-" + stepName + "-source-" + sourceName + "-" + x.Topic
-	config, err := kafka.NewConfig(ctx, kubernetesInterface, namespace, x.Kafka)
+	config, err := kafka.NewConfig(ctx, secretInterface, x.Kafka)
 	if err != nil {
 		return nil, err
 	}
diff --git a/runner/sidecar/source/s3/s3.go b/runner/sidecar/source/s3/s3.go
index 56d1734f..46918697 100644
--- a/runner/sidecar/source/s3/s3.go
+++ b/runner/sidecar/source/s3/s3.go
@@ -13,7 +13,7 @@ import (
 	"io"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/util/workqueue"
 	"net/http"
 	"os"
@@ -36,8 +36,7 @@ type message struct {
 	Path string `json:"path"`
 }
 
-func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace, pipelineName, stepName, sourceName string, x dfv1.S3Source, f source.Func, leadReplica bool) (source.Interface, error) {
-	secretInterface := kubernetesInterface.CoreV1().Secrets(namespace)
+func New(ctx context.Context, secretInterface corev1.SecretInterface, pipelineName, stepName, sourceName string, x dfv1.S3Source, f source.Func, leadReplica bool) (source.Interface, error) {
 	var accessKeyID string
 	{
 		secretName := x.Credentials.AccessKeyID.Name
@@ -75,6 +74,7 @@ func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespac
 	client := s3.New(options)
 	bucket := x.Bucket
 	jobs := workqueue.New()
+	authorization := sharedutil.RandString()
 	if leadReplica {
 		endpoint := "http://" + pipelineName + "-" + stepName + "/sources/" + sourceName
 		logger.Info("starting lead workers", "source", sourceName)
@@ -90,7 +90,12 @@ func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespac
 					func() {
 						defer jobs.Done(item)
 						key := item.(string)
-						resp, err := http.Post(endpoint, "application/octet-stream", bytes.NewBufferString(key))
+						req, err := http.NewRequest("POST", endpoint, bytes.NewBufferString(key))
+						if err != nil {
+							panic(err)
+						}
+						req.Header.Set("Authorization", authorization)
+						resp, err := http.DefaultClient.Do(req)
 						if err != nil {
 							logger.Error(err, "failed to process object", "key", key)
 						} else {
@@ -134,7 +139,7 @@ func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespac
 		}()
 	}
 	return &s3Source{
-		httpsource.New(sourceName, func(ctx context.Context, msg []byte) error {
+		httpsource.New(sourceName, authorization, func(ctx context.Context, msg []byte) error {
 			key := string(msg)
 			output, err := client.GetObject(ctx, &s3.GetObjectInput{Bucket: &bucket, Key: &key})
 			if err != nil {
diff --git a/runner/sidecar/source/stan/stan.go b/runner/sidecar/source/stan/stan.go
index cd732a2f..ffb4d6bc 100644
--- a/runner/sidecar/source/stan/stan.go
+++ b/runner/sidecar/source/stan/stan.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"math/rand"
 	"net/http"
 	"time"
@@ -15,7 +16,6 @@ import (
 	"github.com/nats-io/stan.go"
 	"github.com/nats-io/stan.go/pb"
 	runtimeutil "k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/client-go/kubernetes"
 )
 
 var logger = sharedutil.NewLogger()
@@ -28,7 +28,7 @@ type stanSource struct {
 	queueName         string
 }
 
-func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespace, pipelineName, stepName string, replica int, sourceName string, x dfv1.STAN, f source.Func) (source.Interface, error) {
+func New(ctx context.Context, secretInterface corev1.SecretInterface, pipelineName, stepName string, replica int, sourceName string, x dfv1.STAN, f source.Func) (source.Interface, error) {
 	genClientID := func() string {
 		// In a particular situation, the stan connection status is inconsistent between stan server and client,
 		// the connection is lost from client side, but the server still thinks it's alive. In this case, use
@@ -41,7 +41,7 @@ func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespac
 	var conn *sharedstan.Conn
 	var err error
 	clientID := genClientID()
-	conn, err = sharedstan.ConnectSTAN(ctx, kubernetesInterface, namespace, x, clientID)
+	conn, err = sharedstan.ConnectSTAN(ctx, secretInterface, x, clientID)
 	if err != nil {
 		return nil, err
 	}
@@ -86,7 +86,7 @@ func New(ctx context.Context, kubernetesInterface kubernetes.Interface, namespac
 					_ = sub.Close()
 					logger.Info("stan connection lost, reconnecting...", "source", sourceName)
 					clientID := genClientID()
-					conn, err = sharedstan.ConnectSTAN(ctx, kubernetesInterface, namespace, x, clientID)
+					conn, err = sharedstan.ConnectSTAN(ctx, secretInterface, x, clientID)
 					if err != nil {
 						logger.Error(err, "failed to reconnect", "source", sourceName, "clientID", clientID)
 						continue
diff --git a/runner/sidecar/sources.go b/runner/sidecar/sources.go
index 4fc0327c..c458007a 100644
--- a/runner/sidecar/sources.go
+++ b/runner/sidecar/sources.go
@@ -15,6 +15,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"io"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"time"
 )
 
@@ -32,7 +33,6 @@ func connectSources(ctx context.Context, toMain func(context.Context, []byte) er
 		}
 
 		rateCounter := ratecounter.NewRateCounter(updateInterval)
-		logger.Info("retry config", "source", sourceName, "backoff", s.Retry)
 		f := func(ctx context.Context, msg []byte) error {
 			rateCounter.Incr(1)
 			withLock(func() {
@@ -68,21 +68,26 @@ func connectSources(ctx context.Context, toMain func(context.Context, []byte) er
 				sources[sourceName] = y
 			}
 		} else if x := s.STAN; x != nil {
-			if y, err := stan.New(ctx, kubernetesInterface, namespace, pipelineName, stepName, replica, sourceName, *x, f); err != nil {
+			if y, err := stan.New(ctx, secretInterface, pipelineName, stepName, replica, sourceName, *x, f); err != nil {
 				return err
 			} else {
 				sources[sourceName] = y
 			}
 		} else if x := s.Kafka; x != nil {
-			if y, err := kafkasource.New(ctx, kubernetesInterface, namespace, pipelineName, stepName, sourceName, *x, f); err != nil {
+			if y, err := kafkasource.New(ctx, secretInterface, pipelineName, stepName, sourceName, *x, f); err != nil {
 				return err
 			} else {
 				sources[sourceName] = y
 			}
 		} else if x := s.HTTP; x != nil {
-			sources[sourceName] = httpsource.New(sourceName, f)
+			// we don't want to share this secret
+			secret, err := secretInterface.Get(ctx, step.Name, metav1.GetOptions{})
+			if err != nil {
+				return fmt.Errorf("failed to get secret %q: %w", step.Name, err)
+			}
+			sources[sourceName] = httpsource.New(sourceName, string(secret.Data[fmt.Sprintf("sources.%s.http.authorization", sourceName)]), f)
 		} else if x := s.S3; x != nil {
-			if y, err := s3source.New(ctx, kubernetesInterface, namespace, pipelineName, stepName, sourceName, *x, f, leadReplica()); err != nil {
+			if y, err := s3source.New(ctx, secretInterface, pipelineName, stepName, sourceName, *x, f, leadReplica()); err != nil {
 				return err
 			} else {
 				sources[sourceName] = y
diff --git a/runner/sidecar/stan.go b/runner/sidecar/stan.go
index fee1d27c..0b8c2aa0 100644
--- a/runner/sidecar/stan.go
+++ b/runner/sidecar/stan.go
@@ -9,7 +9,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierr "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
 )
 
 func subjectiveStan(x *dfv1.STAN) {
@@ -46,8 +45,8 @@ func stanFromSecret(s *dfv1.STAN, secret *corev1.Secret) error {
 	return nil
 }
 
-func enrichSTAN(ctx context.Context, secrets v1.SecretInterface, x *dfv1.STAN) error {
-	secret, err := secrets.Get(ctx, "dataflow-stan-"+x.Name, metav1.GetOptions{})
+func enrichSTAN(ctx context.Context, x *dfv1.STAN) error {
+	secret, err := secretInterface.Get(ctx, "dataflow-stan-"+x.Name, metav1.GetOptions{})
 	if err != nil {
 		if !apierr.IsNotFound(err) {
 			return err


Test patch:
diff --git a/runner/sidecar/kafka_test.go b/runner/sidecar/kafka_test.go
index ff2f6397..907e04d4 100644
--- a/runner/sidecar/kafka_test.go
+++ b/runner/sidecar/kafka_test.go
@@ -66,8 +66,9 @@ func Test_kafkaFromSecret(t *testing.T) {
 func Test_enrichKafka(t *testing.T) {
 	t.Run("NotFound", func(t *testing.T) {
 		k := fake.NewSimpleClientset()
+		secretInterface = k.CoreV1().Secrets("")
 		x := &dfv1.Kafka{}
-		err := enrichKafka(context.Background(), k.CoreV1().Secrets(""), x)
+		err := enrichKafka(context.Background(), x)
 		assert.NoError(t, err)
 	})
 	t.Run("Found", func(t *testing.T) {
@@ -79,8 +80,9 @@ func Test_enrichKafka(t *testing.T) {
 				"commitN": []byte("123"),
 			},
 		})
+		secretInterface = k.CoreV1().Secrets("")
 		x := &dfv1.Kafka{Name: "foo"}
-		err := enrichKafka(context.Background(), k.CoreV1().Secrets(""), x)
+		err := enrichKafka(context.Background(), x)
 		assert.NoError(t, err)
 	})
 }
diff --git a/test/http.go b/test/http.go
index 981489d0..1195692b 100644
--- a/test/http.go
+++ b/test/http.go
@@ -13,13 +13,18 @@ import (
 )
 
 func SendMessageViaHTTP(msg string) {
-	r, err := http.Post("http://localhost:3569/sources/default", "text/plain", bytes.NewBufferString(msg))
+	req, err := http.NewRequest("POST", "http://localhost:3569/sources/default", bytes.NewBufferString(msg))
+	if err != nil {
+		panic(err)
+	}
+	req.Header.Set("Authorization", "Bearer my-bearer-token")
+	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		panic(err)
 	} else {
-		body, _ := ioutil.ReadAll(r.Body)
-		if r.StatusCode != 204 {
-			panic(fmt.Errorf("%s: %q", r.Status, body))
+		body, _ := ioutil.ReadAll(resp.Body)
+		if resp.StatusCode != 204 {
+			panic(fmt.Errorf("%s: %q", resp.Status, body))
 		}
 	}
 }
diff --git a/test/pipeline.go b/test/pipeline.go
index fa589043..4f04afb8 100644
--- a/test/pipeline.go
+++ b/test/pipeline.go
@@ -5,6 +5,7 @@ package test
 import (
 	"context"
 	"fmt"
+	corev1 "k8s.io/api/core/v1"
 	"log"
 	"time"
 
@@ -39,12 +40,32 @@ func DeletePipelines() {
 }
 
 func CreatePipeline(pl Pipeline) {
+	ctx := context.Background()
 	log.Printf("creating pipeline %q\n", pl.Name)
 	un := ToUnstructured(pl)
-	_, err := pipelineInterface.Create(context.Background(), un, metav1.CreateOptions{})
+	created, err := pipelineInterface.Create(ctx, un, metav1.CreateOptions{})
 	if err != nil {
 		panic(err)
 	}
+	createSecretsForHTTPSources(pl, FromUnstructured(created), ctx)
+}
+
+func createSecretsForHTTPSources(pl Pipeline, x Pipeline, ctx context.Context) {
+	for _, step := range x.Spec.Steps {
+		for _, source := range step.Sources {
+			if source.HTTP != nil {
+				secretName := fmt.Sprintf("%s-%s", pl.Name, step.Name)
+				log.Printf("creating secret %q\n", secretName)
+				_, err := kubernetesInterface.CoreV1().Secrets(namespace).Create(ctx, &corev1.Secret{
+					ObjectMeta: metav1.ObjectMeta{Name: secretName},
+					StringData: map[string]string{fmt.Sprintf("sources.%s.http.authorization", source.Name): "Bearer my-bearer-token"},
+				}, metav1.CreateOptions{})
+				if sharedutil.IgnoreAlreadyExists(err) != nil {
+					panic(err)
+				}
+			}
+		}
+	}
 }
 
 func WaitForPipeline(opts ...interface{}) {
diff --git a/test/stress/http_stress_test.go b/test/stress/http_stress_test.go
index 3b14fe8c..da0d362b 100644
--- a/test/stress/http_stress_test.go
+++ b/test/stress/http_stress_test.go
@@ -28,6 +28,7 @@ func TestHTTPSourceStress(t *testing.T) {
 	})
 
 	WaitForPipeline()
+	WaitForPod()
 
 	defer StartPortForward("http-main-0")()
 
diff --git a/test/stress/test-results.json b/test/stress/test-results.json
index de1e3b32..138ca257 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": 600,
+  "TestHTTPSourceStress/currentContext=docker-desktop,replicas=1,n=10000.tps": 400,
   "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 fe2f7740..840995ae 100644
--- a/testapi/http.go
+++ b/testapi/http.go
@@ -1,10 +1,10 @@
 package main
 
 import (
+	"bytes"
 	"fmt"
 	"net/http"
 	"strconv"
-	"strings"
 	"time"
 )
 
@@ -38,12 +38,18 @@ func init() {
 		// https://gobyexample.com/worker-pools
 		worker := func(jobs <-chan req, results chan<- interface{}) {
 			for j := range jobs {
-				if resp, err := http.Post(url, "application/octet-stream", strings.NewReader(j.msg)); err != nil {
+				req, err := http.NewRequest("POST", url, bytes.NewBufferString(j.msg))
+				if 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)
+					req.Header.Set("Authorization", "Bearer my-bearer-token")
+					if resp, err := http.DefaultClient.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)
+					}
 				}
 			}
 		}