client

package
v0.8.7 Latest Latest
Warning

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

 Go to latest Published: Jul 2, 2026 License: Apache-2.0 Imports: 35 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	SHA256 scram.HashGeneratorFcn = sha256.New
	SHA512 scram.HashGeneratorFcn = sha512.New
)

Functions

func NewCloudWatchClient 

func NewCloudWatchClient(region string) (*cloudwatch.Client, error)

func NewCostExplorerClient 

func NewCostExplorerClient(region string) (*costexplorer.Client, error)

func NewEC2Client added in v0.2.2 

func NewEC2Client(region string) (*ec2.Client, error)

func NewGlueClient added in v0.7.2 

func NewGlueClient(ctx context.Context, region string) (*glue.Client, error)

func NewIAMClient added in v0.2.1 

func NewIAMClient() (*iam.Client, error)

func NewKafkaClient added in v0.7.0 

func NewKafkaClient(brokerAddresses []string, region string, opts ...AdminOption) (sarama.Client, error)

NewKafkaClient creates a sarama.Client (not a ClusterAdmin) for offset fetching. Uses the same auth configuration options as NewKafkaAdmin.

func NewMSKConnectClient added in v0.4.0 

func NewMSKConnectClient(region string) (*kafkaconnect.Client, error)

func NewS3Client added in v0.2.2 

func NewS3Client(region string) (*s3.Client, error)

func NewSchemaRegistryClient added in v0.4.2 

func NewSchemaRegistryClient(url string, opts ...SchemaRegistryOption) (schemaregistry.Client, error)

NewSchemaRegistryClient creates a new Schema Registry client for the given URL

func ValidateConfluentSchemaRegistryURL added in v0.8.7 

func ValidateConfluentSchemaRegistryURL(rawURL string, opts ...SchemaRegistryOption) error

ValidateConfluentSchemaRegistryURL performs a pre-flight GET against the Schema Registry's /config endpoint (the same endpoint the scan's first call hits) and returns a clear, class-distinct error when the endpoint is not a usable Schema Registry. It honors the same auth options as the scan and returns nil only when the endpoint responds 2xx with a JSON body.

It exists because confluent-kafka-go decodes any 2xx body as JSON without checking Content-Type; pointing --url at an HTML endpoint otherwise surfaces the opaque "invalid character '<' looking for beginning of value".

Types

type AdminConfig 

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

AdminConfig holds the configuration for creating a Kafka admin client

type AdminOption 

type AdminOption func(*AdminConfig)

AdminOption is a function type for configuring the Kafka admin client

func AdminOptionForAuth added in v0.7.0 

func AdminOptionForAuth(authType types.AuthType, clusterAuth types.ClusterAuth) AdminOption

AdminOptionForAuth maps a credential auth type to the corresponding AdminOption. Retained for callers keyed to types.ClusterAuth; delegates to AdminOptionForAuthMethod.

func AdminOptionForAuthMethod added in v0.8.6 

func AdminOptionForAuthMethod(authType types.AuthType, auth types.AuthMethodConfig, skipTLSVerify bool) (AdminOption, error)

AdminOptionForAuthMethod maps an auth type + method config to the corresponding AdminOption. skipTLSVerify applies to SASL/SCRAM (MSK passes false — AWS-managed certs; Apache Kafka passes its InsecureSkipTLSVerify).

func WithIAMAuth 

func WithIAMAuth() AdminOption

WithIAMAuth configures the admin client to use IAM authentication

func WithInsecureSkipVerify added in v0.7.2 

func WithInsecureSkipVerify() AdminOption

WithInsecureSkipVerify disables TLS certificate verification.

func WithSASLPlainAuth added in v0.7.0 

func WithSASLPlainAuth(username, password string) AdminOption

WithSASLPlainAuth configures SASL/PLAIN authentication (used for Confluent Cloud).

func WithSASLPlainAuthNoTLS added in v0.7.2 

func WithSASLPlainAuthNoTLS(username, password string) AdminOption

WithSASLPlainAuthNoTLS configures SASL/PLAIN authentication without TLS encryption. Used for source clusters using SASL_PLAINTEXT listeners.

func WithSASLSCRAMAuth 

func WithSASLSCRAMAuth(username, password, mechanism string, insecureSkipTLSVerify bool) AdminOption

WithSASLSCRAMAuth configures the admin client to use SASL/SCRAM authentication. Set insecureSkipTLSVerify to true only in test environments with self-signed certificates.

func WithTLSAuth 

func WithTLSAuth(caCertFile string, clientCertFile string, clientKeyFile string) AdminOption

func WithUnauthenticatedPlaintextAuth added in v0.4.0 

func WithUnauthenticatedPlaintextAuth() AdminOption

func WithUnauthenticatedTlsAuth added in v0.4.0 

func WithUnauthenticatedTlsAuth() AdminOption

type ClusterKafkaMetadata 

type ClusterKafkaMetadata struct {
	Brokers      []*sarama.Broker
	ControllerID int32
	ClusterID    string
}

ClusterKafkaMetadata represents cluster information including brokers, controller, and cluster ID

type JolokiaClient added in v0.8.0 

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

JolokiaClient is an HTTP client for querying Jolokia REST endpoints

func NewJolokiaClient added in v0.8.0 

func NewJolokiaClient(baseURL string, opts ...JolokiaOption) *JolokiaClient

NewJolokiaClient creates a new Jolokia HTTP client

func (*JolokiaClient) BaseURL added in v0.8.0 

func (c *JolokiaClient) BaseURL() string

BaseURL returns the Jolokia endpoint base URL

func (*JolokiaClient) ReadMBean added in v0.8.0 

func (c *JolokiaClient) ReadMBean(ctx context.Context, mbeanPath string) (map[string]any, error)

ReadMBean queries a Jolokia endpoint for JMX MBean data mbeanPath is the MBean ObjectName (e.g., "kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec") Returns the "value" map from the Jolokia response

func (*JolokiaClient) ReadMBeanAggregate added in v0.8.0 

func (c *JolokiaClient) ReadMBeanAggregate(ctx context.Context, mbeanPattern string, attribute string) (float64, error)

ReadMBeanAggregate queries a wildcard MBean pattern and sums a numeric attribute across all matching MBeans. Used for metrics like log size (per-partition) and connection count (per-listener) that need to be aggregated.

type JolokiaOption added in v0.8.0 

type JolokiaOption func(*JolokiaClient)

JolokiaOption is a functional option for configuring the JolokiaClient

func WithJolokiaBasicAuth added in v0.8.0 

func WithJolokiaBasicAuth(username, password string) JolokiaOption

WithJolokiaBasicAuth configures the client to use HTTP basic authentication

func WithJolokiaTLS added in v0.8.0 

func WithJolokiaTLS(caCertFile string, insecureSkipVerify bool) JolokiaOption

WithJolokiaTLS configures the client to use TLS with an optional custom CA certificate and optional insecure skip verify for self-signed certificates

type KafkaAdmin 

type KafkaAdmin interface {
	ListTopicsWithConfigs() (map[string]sarama.TopicDetail, error)
	GetClusterKafkaMetadata() (*ClusterKafkaMetadata, error)
	DescribeConfig() ([]sarama.ConfigEntry, error)
	ListAcls() ([]sarama.ResourceAcls, error)
	Close() error
}

KafkaAdmin interface defines the Kafka admin operations we need

func NewKafkaAdmin 

func NewKafkaAdmin(brokerAddresses []string, clientBrokerEncryptionInTransit kafkatypes.ClientBroker, region string, kafkaVersion string, opts ...AdminOption) (KafkaAdmin, error)

NewKafkaAdmin creates a new Kafka admin client for the given broker addresses and region

type KafkaAdminClient 

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

KafkaAdminClient wraps sarama.ClusterAdmin to implement our KafkaAdmin interface

func (*KafkaAdminClient) Close 

func (k *KafkaAdminClient) Close() error

func (*KafkaAdminClient) DescribeConfig 

func (k *KafkaAdminClient) DescribeConfig() ([]sarama.ConfigEntry, error)

func (*KafkaAdminClient) GetClusterKafkaMetadata 

func (k *KafkaAdminClient) GetClusterKafkaMetadata() (*ClusterKafkaMetadata, error)

func (*KafkaAdminClient) ListAcls added in v0.1.3 

func (k *KafkaAdminClient) ListAcls() ([]sarama.ResourceAcls, error)

func (*KafkaAdminClient) ListTopicsWithConfigs added in v0.3.4 

func (k *KafkaAdminClient) ListTopicsWithConfigs() (map[string]sarama.TopicDetail, error)

A custom implementation of the ListTopics() function in Sarama that returns all topic configs instead of just overridden configs. This was done to reduce the number of requests to the broker. https://github.com/IBM/sarama/blob/main/admin.go#L349

type MSKAccessTokenProvider 

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

MSKAccessTokenProvider implements sarama.AccessTokenProvider for MSK IAM authentication

func (*MSKAccessTokenProvider) Token 

func (m *MSKAccessTokenProvider) Token() (*sarama.AccessToken, error)

type PrometheusClient added in v0.8.0 

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

PrometheusClient is an HTTP client for querying the Prometheus API

func NewPrometheusClient added in v0.8.0 

func NewPrometheusClient(baseURL string, opts ...PrometheusOption) *PrometheusClient

NewPrometheusClient creates a new Prometheus HTTP client

func (*PrometheusClient) BaseURL added in v0.8.0 

func (c *PrometheusClient) BaseURL() string

BaseURL returns the Prometheus server base URL

func (*PrometheusClient) QueryRange added in v0.8.0 

func (c *PrometheusClient) QueryRange(ctx context.Context, query string, start, end time.Time, step time.Duration) ([]PrometheusMetricResult, error)

QueryRange executes a Prometheus range query and returns parsed results

type PrometheusDataPoint added in v0.8.0 

type PrometheusDataPoint struct {
	Timestamp time.Time
	Value     float64
}

PrometheusDataPoint is a single timestamp+value pair from a Prometheus query

type PrometheusMetricResult added in v0.8.0 

type PrometheusMetricResult struct {
	MetricName string
	Values     []PrometheusDataPoint
}

PrometheusMetricResult holds the parsed result for a single metric from a range query

type PrometheusOption added in v0.8.0 

type PrometheusOption func(*PrometheusClient)

PrometheusOption is a functional option for configuring the PrometheusClient

func WithPrometheusBasicAuth added in v0.8.0 

func WithPrometheusBasicAuth(username, password string) PrometheusOption

WithPrometheusBasicAuth configures the client to use HTTP basic authentication

func WithPrometheusTLS added in v0.8.0 

func WithPrometheusTLS(caCertFile string, insecureSkipVerify bool) PrometheusOption

WithPrometheusTLS configures the client to use TLS with an optional custom CA certificate

type RateLimitedMSKClient added in v0.5.2 

type RateLimitedMSKClient struct {
	*kafka.Client
	// contains filtered or unexported fields
}

func NewMSKClient 

func NewMSKClient(region string, requestsPerSecond float64, burstSize int) (*RateLimitedMSKClient, error)

func (*RateLimitedMSKClient) DescribeTopic added in v0.5.2 

func (c *RateLimitedMSKClient) DescribeTopic(ctx context.Context, params *kafka.DescribeTopicInput, optFns ...func(*kafka.Options)) (*kafka.DescribeTopicOutput, error)

Clusters with high topic count are likely to hit the 429 rate limnit errors. Therefore, we implement an additional layer of retries outside the AWS SDK's standard retryer. If we hit a 429 or quota error, we wait for a new token from our rate limiter and try again.

func (*RateLimitedMSKClient) Wait added in v0.5.2 

func (c *RateLimitedMSKClient) Wait(ctx context.Context) error

type SchemaRegistryConfig added in v0.4.2 

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

SchemaRegistryConfig holds the configuration for creating a Schema Registry client

type SchemaRegistryOption added in v0.4.2 

type SchemaRegistryOption func(*SchemaRegistryConfig)

SchemaRegistryOption is a function type for configuring the Schema Registry client

func WithBasicAuth added in v0.4.2 

func WithBasicAuth(username, password string) SchemaRegistryOption

WithBasicAuth configures the Schema Registry client to use basic authentication

func WithUnauthenticated added in v0.4.2 

func WithUnauthenticated() SchemaRegistryOption

WithUnauthenticated configures the Schema Registry client to use no authentication

type XDGSCRAMClient 

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

func (*XDGSCRAMClient) Begin 

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

func (*XDGSCRAMClient) Done 

func (x *XDGSCRAMClient) Done() bool

func (*XDGSCRAMClient) Step 

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

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.