feat(connection): add HAR collector support to connections

moshloop · web-flow · commit dbb8aeb99cce · 2026-04-03T12:04:05.000Z
* feat(context): add HAR collector context support to AWS connection

Allow HAR collector to be passed through context, falling back to explicit options if not set. This enables middleware collection of HTTP traffic at the context level.

* feat(connection): add HAR collector support to remaining connections

Add HAR collector context support to HTTPConnection, Prometheus,
OpenSearch, GKE, EKS, and CNRM connections. Also add context fallback
to GCS (matching AWS pattern). All connections that make HTTP requests
now support HAR capture via ClientOption or context-level collector.
diff --git a/connection/aws.go b/connection/aws.go
@@ -145,8 +145,12 @@ func (t *AWSConnection) Client(ctx context.Context, opts ...types.ClientOption)
 		TLSClientConfig: &tls.Config{InsecureSkipVerify: t.SkipTLSVerify},
 	}
 
-	if o.HARCollector != nil {
-		tr = o.HARCollector.Middleware()(tr)
+	harCollector := o.HARCollector
+	if harCollector == nil {
+		harCollector = ctx.HARCollector()
+	}
+	if harCollector != nil {
+		tr = harCollector.Middleware()(tr)
 	}
 
 	if ctx.IsTrace() {
diff --git a/connection/cnrm.go b/connection/cnrm.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/flanksource/duty/context"
 	dutyKube "github.com/flanksource/duty/kubernetes"
+	"github.com/flanksource/duty/types"
 	container "google.golang.org/api/container/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -25,8 +26,8 @@ func (t *CNRMConnection) Populate(ctx ConnectionContext) error {
 	return t.GKE.Populate(ctx)
 }
 
-func (t *CNRMConnection) KubernetesClient(ctx context.Context, freshToken bool) (kubernetes.Interface, *rest.Config, error) {
-	cnrmCluster, restConfig, err := t.GKE.KubernetesClient(ctx, freshToken)
+func (t *CNRMConnection) KubernetesClient(ctx context.Context, freshToken bool, opts ...types.ClientOption) (kubernetes.Interface, *rest.Config, error) {
+	cnrmCluster, restConfig, err := t.GKE.KubernetesClient(ctx, freshToken, opts...)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to create Kubernetes client for GKE: %w", err)
 	}
diff --git a/connection/eks.go b/connection/eks.go
@@ -11,6 +11,7 @@ import (
 	signerv4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
 	"github.com/aws/aws-sdk-go-v2/service/eks"
 	"github.com/flanksource/duty/context"
+	"github.com/flanksource/duty/types"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 )
@@ -32,8 +33,8 @@ func (t *EKSConnection) Populate(ctx ConnectionContext) error {
 	return t.AWSConnection.Populate(ctx)
 }
 
-func (t *EKSConnection) KubernetesClient(ctx context.Context, freshToken bool) (kubernetes.Interface, *rest.Config, error) {
-	awsConfig, err := t.AWSConnection.Client(ctx)
+func (t *EKSConnection) KubernetesClient(ctx context.Context, freshToken bool, opts ...types.ClientOption) (kubernetes.Interface, *rest.Config, error) {
+	awsConfig, err := t.AWSConnection.Client(ctx, opts...)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/connection/gcs.go b/connection/gcs.go
@@ -37,12 +37,16 @@ func (g *GCSConnection) Client(ctx context.Context, opts ...types.ClientOption)
 		clientOpts = append(clientOpts, option.WithEndpoint(g.Endpoint))
 	}
 
-	if o.HARCollector != nil {
+	harCollector := o.HARCollector
+	if harCollector == nil {
+		harCollector = ctx.HARCollector()
+	}
+	if harCollector != nil {
 		base := http.RoundTripper(http.DefaultTransport)
 		if g.SkipTLSVerify {
 			base = &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
 		}
-		tr := o.HARCollector.Middleware()(base)
+		tr := harCollector.Middleware()(base)
 		clientOpts = append(clientOpts, option.WithHTTPClient(&http.Client{Transport: tr}))
 	} else if g.SkipTLSVerify {
 		insecureHTTPClient := &http.Client{
diff --git a/connection/gke.go b/connection/gke.go
@@ -7,6 +7,7 @@ import (
 	"net/http"
 
 	"github.com/flanksource/duty/context"
+	"github.com/flanksource/duty/types"
 	"golang.org/x/oauth2/google"
 	container "google.golang.org/api/container/v1"
 	"google.golang.org/api/option"
@@ -35,23 +36,33 @@ func (t *GKEConnection) Validate() *GKEConnection {
 	return t
 }
 
-func (t *GKEConnection) Client(ctx context.Context) (*container.Service, error) {
+func (t *GKEConnection) Client(ctx context.Context, opts ...types.ClientOption) (*container.Service, error) {
 	t = t.Validate()
+	o := types.NewClientOptions(opts...)
 
 	var clientOpts []option.ClientOption
 
 	if t.Endpoint != "" {
 		clientOpts = append(clientOpts, option.WithEndpoint(t.Endpoint))
 	}
 
-	if t.SkipTLSVerify {
-		insecureHTTPClient := &http.Client{
+	harCollector := o.HARCollector
+	if harCollector == nil {
+		harCollector = ctx.HARCollector()
+	}
+	if harCollector != nil {
+		base := http.RoundTripper(http.DefaultTransport)
+		if t.SkipTLSVerify {
+			base = &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}
+		}
+		tr := harCollector.Middleware()(base)
+		clientOpts = append(clientOpts, option.WithHTTPClient(&http.Client{Transport: tr}))
+	} else if t.SkipTLSVerify {
+		clientOpts = append(clientOpts, option.WithHTTPClient(&http.Client{
 			Transport: &http.Transport{
 				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 			},
-		}
-
-		clientOpts = append(clientOpts, option.WithHTTPClient(insecureHTTPClient))
+		}))
 	}
 
 	if t.Credentials != nil && !t.Credentials.IsEmpty() {
@@ -76,8 +87,8 @@ func (t *GKEConnection) Client(ctx context.Context) (*container.Service, error)
 	return svc, nil
 }
 
-func (t *GKEConnection) KubernetesClient(ctx context.Context, freshToken bool) (kubernetes.Interface, *rest.Config, error) {
-	containerService, err := t.Client(ctx)
+func (t *GKEConnection) KubernetesClient(ctx context.Context, freshToken bool, opts ...types.ClientOption) (kubernetes.Interface, *rest.Config, error) {
+	containerService, err := t.Client(ctx, opts...)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/connection/http.go b/connection/http.go
@@ -348,10 +348,15 @@ func (h *HTTPConnection) Hydrate(ctx ConnectionContext, namespace string) (*HTTP
 	return h, nil
 }
 
-func (h HTTPConnection) Transport() netHTTP.RoundTripper {
+func (h HTTPConnection) Transport(opts ...types.ClientOption) netHTTP.RoundTripper {
+	o := types.NewClientOptions(opts...)
+	var base netHTTP.RoundTripper = &netHTTP.Transport{}
+	if o.HARCollector != nil {
+		base = o.HARCollector.Middleware()(base)
+	}
 	rt := &httpConnectionRoundTripper{
 		HTTPConnection: h,
-		Base:           &netHTTP.Transport{},
+		Base:           base,
 	}
 	return rt
 }
@@ -400,8 +405,12 @@ func (rt *httpConnectionRoundTripper) RoundTrip(req *netHTTP.Request) (*netHTTP.
 }
 
 // CreateHTTPClient requires a hydrated connection
-func CreateHTTPClient(ctx ConnectionContext, conn HTTPConnection) (*http.Client, error) {
+func CreateHTTPClient(ctx ConnectionContext, conn HTTPConnection, opts ...types.ClientOption) (*http.Client, error) {
+	o := types.NewClientOptions(opts...)
 	client := http.NewClient()
+	if o.HARCollector != nil {
+		client.HARCollector(o.HARCollector)
+	}
 	if !conn.HTTPBasicAuth.IsEmpty() {
 		client.Auth(conn.GetUsername(), conn.GetPassword())
 		client.Digest(conn.Digest)
diff --git a/connection/kubernetes.go b/connection/kubernetes.go
@@ -176,23 +176,23 @@ func (t KubernetesConnection) Populate(ctx context.Context, freshToken bool, opt
 			return nil, nil, err
 		}
 
-		return t.GKE.KubernetesClient(ctx, freshToken)
+		return t.GKE.KubernetesClient(ctx, freshToken, opts...)
 	}
 
 	if t.EKS != nil {
 		if err := t.EKS.Populate(ctx); err != nil {
 			return nil, nil, err
 		}
 
-		return t.EKS.KubernetesClient(ctx, freshToken)
+		return t.EKS.KubernetesClient(ctx, freshToken, opts...)
 	}
 
 	if t.CNRM != nil {
 		if err := t.CNRM.Populate(ctx); err != nil {
 			return nil, nil, err
 		}
 
-		return t.CNRM.KubernetesClient(ctx, freshToken)
+		return t.CNRM.KubernetesClient(ctx, freshToken, opts...)
 	}
 
 	return nil, nil, nil
diff --git a/connection/opensearch.go b/connection/opensearch.go
@@ -118,7 +118,8 @@ func (c *OpensearchConnection) Hydrate(ctx ConnectionContext) error {
 
 // Client creates and returns an OpenSearch client.
 // NOTE: Must be run on a hydrated OpensearchConnection.
-func (c *OpensearchConnection) Client() (*opensearch.Client, error) {
+func (c *OpensearchConnection) Client(ctx context.Context, opts ...types.ClientOption) (*opensearch.Client, error) {
+	o := types.NewClientOptions(opts...)
 	if len(c.URLs) == 0 {
 		return nil, fmt.Errorf("opensearch connection urls cannot be empty")
 	}
@@ -132,14 +133,28 @@ func (c *OpensearchConnection) Client() (*opensearch.Client, error) {
 		cfg.Password = c.GetPassword()
 	}
 
+	var tr http.RoundTripper
 	if c.InsecureSkipVerify {
-		cfg.Transport = &http.Transport{
-			TLSClientConfig: &tls.Config{
-				InsecureSkipVerify: true,
-			},
+		tr = &http.Transport{
+			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 		}
 	}
 
+	harCollector := o.HARCollector
+	if harCollector == nil {
+		harCollector = ctx.HARCollector()
+	}
+	if harCollector != nil {
+		if tr == nil {
+			tr = http.DefaultTransport
+		}
+		tr = harCollector.Middleware()(tr)
+	}
+
+	if tr != nil {
+		cfg.Transport = tr
+	}
+
 	client, err := opensearch.NewClient(cfg)
 	if err != nil {
 		return nil, fmt.Errorf("error creating opensearch client: %w", err)
@@ -148,7 +163,7 @@ func (c *OpensearchConnection) Client() (*opensearch.Client, error) {
 	return client, nil
 }
 
-func NewOpenSearchClient(ctx context.Context, connection models.Connection) (*opensearch.Client, error) {
+func NewOpenSearchClient(ctx context.Context, connection models.Connection, opts ...types.ClientOption) (*opensearch.Client, error) {
 	var conn OpensearchConnection
 	if err := conn.FromModel(connection); err != nil {
 		return nil, fmt.Errorf("error creating opensearch connection from model: %w", err)
@@ -158,5 +173,5 @@ func NewOpenSearchClient(ctx context.Context, connection models.Connection) (*op
 		return nil, fmt.Errorf("error hydrating opensearch connection: %w", err)
 	}
 
-	return conn.Client()
+	return conn.Client(ctx, opts...)
 }
diff --git a/connection/prometheus.go b/connection/prometheus.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/flanksource/duty/context"
 	"github.com/flanksource/duty/models"
+	"github.com/flanksource/duty/types"
 )
 
 // +kubebuilder:object:generate=true
@@ -34,10 +35,10 @@ func (p *PrometheusConnection) Populate(ctx ConnectionContext) error {
 	return nil
 }
 
-func (p *PrometheusConnection) NewClient(ctx context.Context) (v1.API, error) {
+func (p *PrometheusConnection) NewClient(ctx context.Context, opts ...types.ClientOption) (v1.API, error) {
 	cfg := api.Config{
 		Address:      p.HTTPConnection.URL,
-		RoundTripper: p.HTTPConnection.Transport(),
+		RoundTripper: p.HTTPConnection.Transport(opts...),
 	}
 
 	client, err := api.NewClient(cfg)
diff --git a/context/context.go b/context/context.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	commons "github.com/flanksource/commons/context"
+	"github.com/flanksource/commons/har"
 	"github.com/flanksource/commons/logger"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/labstack/echo/v4"
@@ -313,6 +314,17 @@ func (k Context) WithNamespace(namespace string) Context {
 	return k.WithValue("namespace", namespace)
 }
 
+func (k Context) WithHARCollector(collector *har.Collector) Context {
+	return k.WithValue("har-collector", collector)
+}
+
+func (k Context) HARCollector() *har.Collector {
+	if v, ok := k.Value("har-collector").(*har.Collector); ok {
+		return v
+	}
+	return nil
+}
+
 func (k Context) WithDB(db *gorm.DB, pool *pgxpool.Pool) Context {
 	return k.WithValue("db", db).WithValue("pgxpool", pool)
 }
diff --git a/go.sum b/go.sum
diff --git a/logs/opensearch/search.go b/logs/opensearch/search.go

Original file line number	Diff line number	Diff line change
`@@ -145,8 +145,12 @@ func (t *AWSConnection) Client(ctx context.Context, opts ...types.ClientOption)`
`145`	`145`	`TLSClientConfig: &tls.Config{InsecureSkipVerify: t.SkipTLSVerify},`
`146`	`146`	`}`
`147`	`147`
`148`		`- if o.HARCollector != nil {`
`149`		`- tr = o.HARCollector.Middleware()(tr)`
	`148`	`+ harCollector := o.HARCollector`
	`149`	`+ if harCollector == nil {`
	`150`	`+ harCollector = ctx.HARCollector()`
	`151`	`+ }`
	`152`	`+ if harCollector != nil {`
	`153`	`+ tr = harCollector.Middleware()(tr)`
`150`	`154`	`}`
`151`	`155`
`152`	`156`	`if ctx.IsTrace() {`
Original file line number	Diff line number	Diff line change
`@@ -11,6 +11,7 @@ import (`
`11`	`11`	`signerv4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"`
`12`	`12`	`"github.com/aws/aws-sdk-go-v2/service/eks"`
`13`	`13`	`"github.com/flanksource/duty/context"`
	`14`	`+ "github.com/flanksource/duty/types"`
`14`	`15`	`"k8s.io/client-go/kubernetes"`
`15`	`16`	`"k8s.io/client-go/rest"`
`16`	`17`	`)`
`@@ -32,8 +33,8 @@ func (t *EKSConnection) Populate(ctx ConnectionContext) error {`
`32`	`33`	`return t.AWSConnection.Populate(ctx)`
`33`	`34`	`}`
`34`	`35`
`35`		`-func (t EKSConnection) KubernetesClient(ctx context.Context, freshToken bool) (kubernetes.Interface, rest.Config, error) {`
`36`		`- awsConfig, err := t.AWSConnection.Client(ctx)`
	`36`	`+func (t EKSConnection) KubernetesClient(ctx context.Context, freshToken bool, opts ...types.ClientOption) (kubernetes.Interface, rest.Config, error) {`
	`37`	`+ awsConfig, err := t.AWSConnection.Client(ctx, opts...)`
`37`	`38`	`if err != nil {`
`38`	`39`	`return nil, nil, err`
`39`	`40`	`}`
Original file line number	Diff line number	Diff line change
`@@ -176,23 +176,23 @@ func (t KubernetesConnection) Populate(ctx context.Context, freshToken bool, opt`
`176`	`176`	`return nil, nil, err`
`177`	`177`	`}`
`178`	`178`
`179`		`- return t.GKE.KubernetesClient(ctx, freshToken)`
	`179`	`+ return t.GKE.KubernetesClient(ctx, freshToken, opts...)`
`180`	`180`	`}`
`181`	`181`
`182`	`182`	`if t.EKS != nil {`
`183`	`183`	`if err := t.EKS.Populate(ctx); err != nil {`
`184`	`184`	`return nil, nil, err`
`185`	`185`	`}`
`186`	`186`
`187`		`- return t.EKS.KubernetesClient(ctx, freshToken)`
	`187`	`+ return t.EKS.KubernetesClient(ctx, freshToken, opts...)`
`188`	`188`	`}`
`189`	`189`
`190`	`190`	`if t.CNRM != nil {`
`191`	`191`	`if err := t.CNRM.Populate(ctx); err != nil {`
`192`	`192`	`return nil, nil, err`
`193`	`193`	`}`
`194`	`194`
`195`		`- return t.CNRM.KubernetesClient(ctx, freshToken)`
	`195`	`+ return t.CNRM.KubernetesClient(ctx, freshToken, opts...)`
`196`	`196`	`}`
`197`	`197`
`198`	`198`	`return nil, nil, nil`
Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@ import (`
`8`	`8`
`9`	`9`	`"github.com/flanksource/duty/context"`
`10`	`10`	`"github.com/flanksource/duty/models"`
	`11`	`+ "github.com/flanksource/duty/types"`
`11`	`12`	`)`
`12`	`13`
`13`	`14`	`// +kubebuilder:object:generate=true`
`@@ -34,10 +35,10 @@ func (p *PrometheusConnection) Populate(ctx ConnectionContext) error {`
`34`	`35`	`return nil`
`35`	`36`	`}`
`36`	`37`
`37`		`-func (p *PrometheusConnection) NewClient(ctx context.Context) (v1.API, error) {`
	`38`	`+func (p *PrometheusConnection) NewClient(ctx context.Context, opts ...types.ClientOption) (v1.API, error) {`
`38`	`39`	`cfg := api.Config{`
`39`	`40`	`Address: p.HTTPConnection.URL,`
`40`		`- RoundTripper: p.HTTPConnection.Transport(),`
	`41`	`+ RoundTripper: p.HTTPConnection.Transport(opts...),`
`41`	`42`	`}`
`42`	`43`
`43`	`44`	`client, err := api.NewClient(cfg)`