transport

package
v0.1.12 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 1, 2026 License: Apache-2.0 Imports: 24 Imported by: 0

Documentation

Overview

internal/transport/dakr_client.go

internal/transport/interface.go

internal/transport/percentile_fetcher.go

internal/transport/sender.go

internal/transport/snapshot_batched.go

internal/transport/telemetry_sender.go

Index

Constants

View Source 
const (
	DefaultOperatorType = "zxporter"

	// Header keys for client identification in transport layer
	HeaderClient          = "X-Client"
	HeaderOperatorType    = "X-Operator-Type"
	HeaderOperatorVersion = "X-Operator-Version"
	HeaderOperatorGitSHA  = "X-Operator-Git-SHA"
)

Variables

View Source 
var ErrSnapshotBatchedUnsupported = errors.New("dakr does not support SendClusterSnapshotBatched")

ErrSnapshotBatchedUnsupported signals that the dakr backend does not serve SendClusterSnapshotBatched yet; callers fall back to the legacy SendClusterSnapshotStream path.

Functions

This section is empty.

Types

type ClientHeaders added in v0.0.63 

type ClientHeaders struct {
	// contains filtered or unexported fields
}

ClientHeaders holds the operator identification information for transport headers. This struct is used by the interceptor to access current client information.

func NewClientHeaders added in v0.0.63 

func NewClientHeaders(clusterToken string) *ClientHeaders

NewClientHeaders creates a new ClientHeaders instance

func (*ClientHeaders) AttachToRequest added in v0.0.63 

func (h *ClientHeaders) AttachToRequest(header http.Header)

AttachToRequest adds all client identification headers to a Connect request. This is the core method that sets: - Authorization: Bearer token for authentication - X-Client: Combined operator type and version (e.g., "zxporter/1.0.0") - X-Operator-Type: The operator type identifier - X-Operator-Version: The operator version - X-Operator-Git-SHA: The git commit SHA of the operator build

func (*ClientHeaders) GetClusterToken added in v0.0.63 

func (h *ClientHeaders) GetClusterToken() string

GetClusterToken returns the current cluster token (thread-safe)

func (*ClientHeaders) SetClusterToken added in v0.0.63 

func (h *ClientHeaders) SetClusterToken(token string)

SetClusterToken updates the cluster token (thread-safe)

type DakrClient 

type DakrClient interface {
	// SendResource sends any resource data to Dakr
	SendResource(ctx context.Context, resource collector.CollectedResource) (string, error)
	// SendResourceBatch sends a batch of resources of the same type to Dakr
	SendResourceBatch(
		ctx context.Context,
		resources []collector.CollectedResource,
		resourceType collector.ResourceType,
	) (string, error)
	// SendTelemetryMetrics sends telemetry metrics to Dakr
	SendTelemetryMetrics(ctx context.Context, metrics []*dto.MetricFamily) (int32, error)
	// SendClusterSnapshotStream sends cluster snapshot data via streaming for large payloads
	SendClusterSnapshotStream(
		ctx context.Context,
		snapshot *gen.ClusterSnapshot,
		snapshotID string,
		timestamp time.Time,
	) (string, *gen.ClusterSnapshot, error)
	// OpenClusterSnapshotBatchStream opens a per-resource-type batched
	// snapshot stream (memory-bounded successor of SendClusterSnapshotStream).
	OpenClusterSnapshotBatchStream(
		ctx context.Context,
		snapshotID string,
		timestamp time.Time,
		isFullSnapshot bool,
	) (SnapshotBatchStream, error)
	// telemetry_logger.TelemetryLogSender sends a batch of log entries to Dakr
	telemetry_logger.TelemetryLogSender
	// ExchangePATForClusterToken exchanges a PAT token for a cluster token
	ExchangePATForClusterToken(
		ctx context.Context,
		patToken, clusterName, k8sProvider string,
	) (string, string, error)

	// SendNetworkTrafficMetrics pushes network traffic metrics from a node
	SendNetworkTrafficMetrics(
		ctx context.Context,
		req *gen.SendNetworkTrafficMetricsRequest,
	) (*gen.SendNetworkTrafficMetricsResponse, error)

	// SendScalerMetrics pushes a batch of KEDA scaler metric samples from a cluster.
	SendScalerMetrics(
		ctx context.Context,
		clusterID string,
		teamID string,
		samples []*gen.ScalerMetricSample,
	) error

	// SendContainerRedMetrics pushes per-container inbound RED metrics from a node.
	// It is a separate call from SendNetworkTrafficMetrics rather than a field on it
	// because the two carry different grains -- flows between endpoints versus
	// per-container per-minute aggregates -- and because a failure to deliver one
	// must not discard the other.
	SendContainerRedMetrics(
		ctx context.Context,
		req *gen.SendContainerRedMetricsRequest,
	) (*gen.SendContainerRedMetricsResponse, error)

	// ReportHealth
	ReportHealth(ctx context.Context, req *gen.ReportHealthRequest) error

	// NewPercentileFetcher creates a PercentileFetcher that retrieves
	// pre-computed workload percentiles from the DAKR control plane.
	NewPercentileFetcher() *DakrPercentileFetcher
}

DakrClient defines methods for sending data to Dakr

func NewDakrClient 

func NewDakrClient(dakrBaseURL string, clusterToken string, logger logr.Logger) DakrClient

NewDakrClient creates a new client for Dakr service

func NewSimpleDakrClient 

func NewSimpleDakrClient(logger logr.Logger) DakrClient

NewSimpleDakrClient creates a new simple Dakr client for development/testing

type DakrPercentileFetcher added in v0.0.82 

type DakrPercentileFetcher struct {
	// contains filtered or unexported fields
}

DakrPercentileFetcher implements collector.PercentileFetcher using the DAKR control plane's GetWorkloadContainerPercentiles RPC.

func NewDakrPercentileFetcher added in v0.0.82 

func NewDakrPercentileFetcher(
	client genconnect.K8SServiceClient,
	headers *ClientHeaders,
	logger logr.Logger,
) *DakrPercentileFetcher

NewDakrPercentileFetcher creates a new fetcher that calls the DAKR control plane to retrieve pre-computed percentiles for workload containers.

func (*DakrPercentileFetcher) FetchWorkloadPercentiles added in v0.0.82 

func (f *DakrPercentileFetcher) FetchWorkloadPercentiles(
	ctx context.Context,
	clusterID string,
	workloads []collector.HistoricalWorkloadQuery,
) (map[string]*gen.HistoricalMetricsSummary, error)

FetchWorkloadPercentiles calls the DAKR control plane for each workload and maps the response into gen.HistoricalMetricsSummary entries keyed by "namespace/workloadName/workloadKind".

type DirectDakrSender 

type DirectDakrSender struct {
	// contains filtered or unexported fields
}

DirectDakrSender sends resources directly to Dakr without buffering

func (*DirectDakrSender) OpenClusterSnapshotBatchStream added in v0.1.0 

func (s *DirectDakrSender) OpenClusterSnapshotBatchStream(
	ctx context.Context,
	snapshotID string,
	timestamp time.Time,
	isFullSnapshot bool,
) (SnapshotBatchStream, error)

OpenClusterSnapshotBatchStream injects the sender's cluster/team identity and opens a batched snapshot stream on the underlying client.

func (*DirectDakrSender) Send 

func (s *DirectDakrSender) Send(
	ctx context.Context,
	resource collector.CollectedResource,
) (string, error)

Send transmits a resource directly to Dakr

func (*DirectDakrSender) SendBatch 

func (s *DirectDakrSender) SendBatch(
	ctx context.Context,
	resources []collector.CollectedResource,
	resourceType collector.ResourceType,
) (string, error)

SendBatch transmits a batch of resources of the same type to Dakr

func (*DirectDakrSender) SendClusterSnapshotStream 

func (s *DirectDakrSender) SendClusterSnapshotStream(
	ctx context.Context,
	snapshot *gen.ClusterSnapshot,
	snapshotID string,
	timestamp time.Time,
) (string, *gen.ClusterSnapshot, error)

Update sender.go with fallback logic

func (*DirectDakrSender) SetClusterID 

func (s *DirectDakrSender) SetClusterID(clusterID string)

func (*DirectDakrSender) SetTeamID 

func (s *DirectDakrSender) SetTeamID(teamID string)

func (*DirectDakrSender) Start 

func (s *DirectDakrSender) Start(ctx context.Context) error

Start initializes the sender (no-op for direct sender)

func (*DirectDakrSender) Stop 

func (s *DirectDakrSender) Stop() error

Stop cleans up resources (no-op for direct sender)

type DirectSender 

type DirectSender interface {
	// SendBatch transmits a batch of resources of the same type to the target system
	SendBatch(
		ctx context.Context,
		resource []collector.CollectedResource,
		resourceType collector.ResourceType,
	) (string, error)

	// Send transmits a resource to the target system
	Send(ctx context.Context, resource collector.CollectedResource) (string, error)

	// SendClusterSnapshotStream sends large cluster snapshot data using streaming
	SendClusterSnapshotStream(
		ctx context.Context,
		snapshot *gen.ClusterSnapshot,
		snapshotID string,
		timestamp time.Time,
	) (string, *gen.ClusterSnapshot, error)

	// telemetry_logger.TelemetryLogSender sends a batch of log entries.
	telemetry_logger.TelemetryLogSender
}

DirectSender defines methods for sending data to external systems directly

func NewDirectSender 

func NewDirectSender(dakrClient DakrClient, logger logr.Logger) DirectSender

NewDirectSender creates a new DirectSender. It expects a non-nil DakrClient.

type RealDakrClient 

type RealDakrClient struct {
	// contains filtered or unexported fields
}

RealDakrClient implements communication with Dakr service

func (*RealDakrClient) ExchangePATForClusterToken 

func (c *RealDakrClient) ExchangePATForClusterToken(
	ctx context.Context,
	patToken, clusterName, k8sProvider string,
) (string, string, error)

ExchangePATForClusterToken exchanges a PAT token for a cluster token

func (*RealDakrClient) NewPercentileFetcher added in v0.0.82 

func (c *RealDakrClient) NewPercentileFetcher() *DakrPercentileFetcher

NewPercentileFetcher creates a DakrPercentileFetcher that uses the DAKR control plane to retrieve pre-computed workload percentiles, replacing the local Prometheus dependency for historical metrics.

func (*RealDakrClient) OpenClusterSnapshotBatchStream added in v0.1.0 

func (c *RealDakrClient) OpenClusterSnapshotBatchStream(
	ctx context.Context,
	snapshotID string,
	timestamp time.Time,
	isFullSnapshot bool,
) (SnapshotBatchStream, error)

OpenClusterSnapshotBatchStream opens a batched snapshot stream to dakr. Opening is lazy: an unsupported backend surfaces as ErrSnapshotBatchedUnsupported from SendBatch/Finish, not from here.

func (*RealDakrClient) ReportHealth added in v0.0.64 

func (c *RealDakrClient) ReportHealth(ctx context.Context, req *gen.ReportHealthRequest) error

ReportHealth reports the health status of the operator to Dakr

func (*RealDakrClient) SendClusterSnapshotStream 

func (c *RealDakrClient) SendClusterSnapshotStream(
	ctx context.Context,
	snapshot *gen.ClusterSnapshot,
	snapshotID string,
	timestamp time.Time,
) (string, *gen.ClusterSnapshot, error)

SendClusterSnapshotStream sends cluster snapshot data in chunks via streaming

func (*RealDakrClient) SendContainerRedMetrics added in v0.1.12 

func (c *RealDakrClient) SendContainerRedMetrics(
	ctx context.Context,
	req *gen.SendContainerRedMetricsRequest,
) (*gen.SendContainerRedMetricsResponse, error)

SendContainerRedMetrics pushes per-container inbound RED metrics from a node.

There is no retry here, and there must not be one. The items are per-minute deltas destined for a SummingMergeTree, so a re-sent batch is added to the one that already landed and is indistinguishable from real extra traffic. Losing a minute is recoverable by looking at the next one; double counting one is not recoverable at all.

That is enforced by noRetryProcedures rather than by this comment: the client's retry interceptor applies to every procedure it wraps, so the exemption has to be registered there. TestSendContainerRedMetricsIsNeverRetried is what fails if the registration is dropped.

func (*RealDakrClient) SendNetworkTrafficMetrics 

func (c *RealDakrClient) SendNetworkTrafficMetrics(
	ctx context.Context,
	req *gen.SendNetworkTrafficMetricsRequest,
) (*gen.SendNetworkTrafficMetricsResponse, error)

SendNetworkTrafficMetrics pushes network traffic metrics from a node

func (*RealDakrClient) SendResource 

func (c *RealDakrClient) SendResource(
	ctx context.Context,
	resource collector.CollectedResource,
) (string, error)

SendResource sends the resource to Dakr through gRPC

func (*RealDakrClient) SendResourceBatch 

func (c *RealDakrClient) SendResourceBatch(
	ctx context.Context,
	resources []collector.CollectedResource,
	resourceType collector.ResourceType,
) (string, error)

SendResourceBatch sends a batch of resources to Dakr through gRPC

func (*RealDakrClient) SendScalerMetrics added in v0.1.12 

func (c *RealDakrClient) SendScalerMetrics(
	ctx context.Context,
	clusterID string,
	teamID string,
	samples []*gen.ScalerMetricSample,
) error

SendScalerMetrics pushes a batch of KEDA scaler metric samples to Dakr. If clusterID or teamID are empty the values are resolved from the context ("cluster_id" / "team_id" keys) so callers that don't carry them at construction time can still pass "" and rely on context injection.

Note: in production the manager starts collectors with a bare context, so the context path above is rarely populated. The authoritative cluster_id and team_id are set server-side by dakr's ClusterTokenAuth middleware from the Bearer token — the same pattern used by SendNetworkTrafficMetrics. Do not rely on the context values being present; they are a convenience fallback only.

func (*RealDakrClient) SendTelemetryLogs 

func (c *RealDakrClient) SendTelemetryLogs(
	ctx context.Context,
	in *gen.SendTelemetryLogsRequest,
) (*gen.SendTelemetryLogsResponse, error)

SendTelemetryLogs sends a batch of log entries to the Dakr service.

func (*RealDakrClient) SendTelemetryMetrics 

func (c *RealDakrClient) SendTelemetryMetrics(
	ctx context.Context,
	metrics []*dto.MetricFamily,
) (int32, error)

SendTelemetryMetrics sends telemetry metrics to Dakr

type RetryPolicy 

type RetryPolicy struct {
	MaxAttempts    int
	InitialBackoff time.Duration
	MaxBackoff     time.Duration
	IsRetryable    func(err error) bool
}

RetryPolicy defines the parameters for retrying.

type Sender 

type Sender interface {
	// Send transmits a resource to the target system
	Send(ctx context.Context, resource collector.CollectedResource) (string, error)

	// Start initializes the sender (establishing connections, etc.)
	Start(ctx context.Context) error

	// Stop cleans up resources
	Stop() error
}

Sender defines methods for sending data to external systems

func NewDirectDakrSender 

func NewDirectDakrSender(dakrClient DakrClient, logger logr.Logger) Sender

type SimpleDakrClient 

type SimpleDakrClient struct {
	// contains filtered or unexported fields
}

SimpleDakrClient is a placeholder implementation of DakrClient

func (*SimpleDakrClient) ExchangePATForClusterToken 

func (c *SimpleDakrClient) ExchangePATForClusterToken(
	ctx context.Context,
	patToken, clusterName, k8sProvider string,
) (string, string, error)

ExchangePATForClusterToken implements SimpleDakrClient.

func (*SimpleDakrClient) NewPercentileFetcher added in v0.0.82 

func (c *SimpleDakrClient) NewPercentileFetcher() *DakrPercentileFetcher

NewPercentileFetcher returns nil for the simple client since it has no backing K8S service connection. Callers must nil-check before use.

func (*SimpleDakrClient) OpenClusterSnapshotBatchStream added in v0.1.0 

func (c *SimpleDakrClient) OpenClusterSnapshotBatchStream(
	ctx context.Context,
	snapshotID string,
	timestamp time.Time,
	isFullSnapshot bool,
) (SnapshotBatchStream, error)

OpenClusterSnapshotBatchStream is unsupported in the simple client; callers fall back to the legacy snapshot path.

func (*SimpleDakrClient) ReportHealth added in v0.0.64 

func (c *SimpleDakrClient) ReportHealth(ctx context.Context, req *gen.ReportHealthRequest) error

ReportHealth implements SimpleDakrClient.

func (*SimpleDakrClient) SendClusterSnapshotStream 

func (c *SimpleDakrClient) SendClusterSnapshotStream(
	ctx context.Context,
	snapshot *gen.ClusterSnapshot,
	snapshotID string,
	timestamp time.Time,
) (string, *gen.ClusterSnapshot, error)

Update sender.go with fallback logic

func (*SimpleDakrClient) SendContainerRedMetrics added in v0.1.12 

func (c *SimpleDakrClient) SendContainerRedMetrics(
	ctx context.Context,
	req *gen.SendContainerRedMetricsRequest,
) (*gen.SendContainerRedMetricsResponse, error)

SendContainerRedMetrics implements SimpleDakrClient.

func (*SimpleDakrClient) SendNetworkTrafficMetrics 

func (c *SimpleDakrClient) SendNetworkTrafficMetrics(
	ctx context.Context,
	req *gen.SendNetworkTrafficMetricsRequest,
) (*gen.SendNetworkTrafficMetricsResponse, error)

SendNetworkTrafficMetrics implements SimpleDakrClient.

func (*SimpleDakrClient) SendResource 

func (c *SimpleDakrClient) SendResource(
	ctx context.Context,
	resource collector.CollectedResource,
) (string, error)

SendResource logs the resource data (for development/testing)

func (*SimpleDakrClient) SendResourceBatch 

func (c *SimpleDakrClient) SendResourceBatch(
	ctx context.Context,
	resources []collector.CollectedResource,
	resourceType collector.ResourceType,
) (string, error)

SendResourceBatch logs the batch resource data (for development/testing)

func (*SimpleDakrClient) SendScalerMetrics added in v0.1.12 

func (c *SimpleDakrClient) SendScalerMetrics(
	ctx context.Context,
	clusterID string,
	teamID string,
	samples []*gen.ScalerMetricSample,
) error

SendScalerMetrics implements SimpleDakrClient (no-op).

func (*SimpleDakrClient) SendTelemetryLogs 

func (c *SimpleDakrClient) SendTelemetryLogs(
	ctx context.Context,
	in *gen.SendTelemetryLogsRequest,
) (*gen.SendTelemetryLogsResponse, error)

SendTelemetryLogs implements SimpleDakrClient.

func (*SimpleDakrClient) SendTelemetryMetrics 

func (c *SimpleDakrClient) SendTelemetryMetrics(
	ctx context.Context,
	metrics []*dto.MetricFamily,
) (int32, error)

SendTelemetryMetrics logs the telemetry metrics data (for development/testing)

type SnapshotBatchStream added in v0.1.0 

type SnapshotBatchStream interface {
	// SendBatch sends one batch for a resource type; typeComplete marks the
	// last batch of that type.
	SendBatch(rt gen.ResourceType, entries []*gen.SnapshotEntry, typeComplete bool) error
	// Finish sends the snapshot_complete marker and returns the response.
	Finish() (*gen.SendClusterSnapshotBatchedResponse, error)
	// Abort closes the stream without the snapshot_complete marker, telling
	// the receiver to treat the snapshot as incomplete. Idempotent.
	Abort()
}

SnapshotBatchStream sends one snapshot's batches to dakr. It stamps the snapshot identity (cluster, team, snapshot ID, timestamp) onto every wire message so callers only provide per-batch content.

type TelemetrySender 

type TelemetrySender struct {
	// contains filtered or unexported fields
}

TelemetrySender is responsible for periodically sending metrics to the DAKR server

func NewTelemetrySender 

func NewTelemetrySender(
	logger logr.Logger,
	dakrClient DakrClient,
	metrics *collector.TelemetryMetrics,
	interval time.Duration,
	healthManager *health.HealthManager,
) *TelemetrySender

NewTelemetrySender creates a new TelemetrySender

func (*TelemetrySender) Start 

func (s *TelemetrySender) Start(ctx context.Context) error

Start begins the periodic sending of metrics

func (*TelemetrySender) Stop 

func (s *TelemetrySender) Stop() error

Stop halts the periodic sending of metrics

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.