Skip to content

Commit

Permalink
lint
Browse files Browse the repository at this point in the history
  • Loading branch information
@benclive
benclive committed Aug 29, 2024
1 parent a51f6e9 commit 72d8af1
Show file tree
Hide file tree
Showing 5 changed files with 49 additions and 49 deletions.
14 changes: 7 additions & 7 deletions pkg/ingester-rf1/ingester.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ var (
compressionStats = analytics.NewString("ingester_compression")
targetSizeStats = analytics.NewInt("ingester_target_size_bytes")
activeTenantsStats = analytics.NewInt("ingester_active_tenants")
ingesterIdRegexp = regexp.MustCompile("ingester(-rf1)-([0-9]+)")
ingesterIDRegexp = regexp.MustCompile("ingester(-rf1)-([0-9]+)")
)

// Config for an ingester.
Expand Down Expand Up @@ -107,7 +107,7 @@ type Config struct {
factory ring_client.PoolFactory `yaml:"-"`

// Used for the kafka ingestion path
PartitionRingConfig partitionring.PartitionRingConfig `yaml:"partition_ring" category:"experimental"`
PartitionRingConfig partitionring.Config `yaml:"partition_ring" category:"experimental"`
KafkaConfig kafka.Config
}

Expand Down Expand Up @@ -225,7 +225,7 @@ type Ingester struct {
customStreamsTracker push.UsageTracker

readRing ring.ReadRing
ingestPartitionId int32
ingesterPartitionID int32
partitionRingLifecycler *ring.PartitionInstanceLifecycler
}

Expand Down Expand Up @@ -283,7 +283,7 @@ func New(cfg Config, clientConfig client.Config,
}

if cfg.KafkaConfig.Enabled {
i.ingestPartitionId, err = ingesterPartitionID(cfg.LifecyclerConfig.ID)
i.ingesterPartitionID, err = ingesterPartitionID(cfg.LifecyclerConfig.ID)
if err != nil {
return nil, fmt.Errorf("calculating ingester partition ID: %w", err)
}
Expand All @@ -297,7 +297,7 @@ func New(cfg Config, clientConfig client.Config,
}

i.partitionRingLifecycler = ring.NewPartitionInstanceLifecycler(
cfg.PartitionRingConfig.ToLifecyclerConfig(i.ingestPartitionId, cfg.LifecyclerConfig.ID),
cfg.PartitionRingConfig.ToLifecyclerConfig(i.ingesterPartitionID, cfg.LifecyclerConfig.ID),
"partition-ring",
"partition-ring-key",
partitionRingKV,
Expand Down Expand Up @@ -336,9 +336,9 @@ func ingesterPartitionID(ingesterID string) (int32, error) {
return 0, nil
}

match := ingesterIdRegexp.FindStringSubmatch(ingesterID)
match := ingesterIDRegexp.FindStringSubmatch(ingesterID)
if len(match) == 0 {
return 0, fmt.Errorf("ingester ID %s doesn't match regular expression %q", ingesterID, ingesterIdRegexp.String())
return 0, fmt.Errorf("ingester ID %s doesn't match regular expression %q", ingesterID, ingesterIDRegexp.String())
}
// Parse the ingester sequence number.
ingesterSeq, err := strconv.Atoi(match[1])
Expand Down
12 changes: 6 additions & 6 deletions pkg/ingester-rf1/kafka/kafka_tee.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
f.BoolVar(&cfg.Enabled, "enabled", false, "enables the use of the Kafka ingest path")
}

type KafkaTee struct {
type Tee struct {
cfg Config
logger log.Logger
kafkaClient *kgo.Client
Expand All @@ -42,13 +42,13 @@ type KafkaTee struct {
ingesterAppends *prometheus.CounterVec
}

func NewKafkaTee(
func NewTee(
cfg Config,
metricsNamespace string,
registerer prometheus.Registerer,
logger log.Logger,
partitionRing *ring.PartitionInstanceRing,
) (*KafkaTee, error) {
) (*Tee, error) {
registerer = prometheus.WrapRegistererWithPrefix(metricsNamespace+"_", registerer)

metrics := kprom.NewMetrics(
Expand Down Expand Up @@ -112,7 +112,7 @@ func NewKafkaTee(
panic("failed to start kafka client")
}

t := &KafkaTee{
t := &Tee{
logger: log.With(logger, "component", "kafka-tee"),
ingesterAppends: promauto.With(registerer).NewCounterVec(prometheus.CounterOpts{
Name: "kafka_ingester_appends_total",
Expand All @@ -127,7 +127,7 @@ func NewKafkaTee(
}

// Duplicate Implements distributor.Tee which is used to tee distributor requests to pattern ingesters.
func (t *KafkaTee) Duplicate(tenant string, streams []distributor.KeyedStream) {
func (t *Tee) Duplicate(tenant string, streams []distributor.KeyedStream) {
for idx := range streams {
go func(stream distributor.KeyedStream) {
if err := t.sendStream(tenant, stream); err != nil {
Expand All @@ -137,7 +137,7 @@ func (t *KafkaTee) Duplicate(tenant string, streams []distributor.KeyedStream) {
}
}

func (t *KafkaTee) sendStream(tenant string, stream distributor.KeyedStream) error {
func (t *Tee) sendStream(tenant string, stream distributor.KeyedStream) error {
partitionID, err := t.partitionRing.PartitionRing().ActivePartitionForKey(stream.HashKey)
if err != nil {
t.ingesterAppends.WithLabelValues("partition_unknown", "fail").Inc()
Expand Down
6 changes: 3 additions & 3 deletions pkg/ingester-rf1/partitionring/partition_ring.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import (
"github.com/grafana/dskit/ring"
)

type PartitionRingConfig struct {
type Config struct {
KVStore kv.Config `yaml:"kvstore" doc:"description=The key-value store used to share the hash ring across multiple instances. This option needs be set on ingesters, distributors, queriers, and rulers when running in microservices mode."`

// MinOwnersCount maps to ring.PartitionInstanceLifecyclerConfig's WaitOwnersCountOnPending.
Expand All @@ -25,7 +25,7 @@ type PartitionRingConfig struct {
}

// RegisterFlags adds the flags required to config this to the given FlagSet
func (cfg *PartitionRingConfig) RegisterFlags(f *flag.FlagSet) {
func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
// Ring flags
cfg.KVStore.Store = "memberlist" // Override default value.
cfg.KVStore.RegisterFlagsWithPrefix("ingester.partition-ring.", "collectors/", f)
Expand All @@ -35,7 +35,7 @@ func (cfg *PartitionRingConfig) RegisterFlags(f *flag.FlagSet) {
f.DurationVar(&cfg.DeleteInactivePartitionAfter, "ingester.partition-ring.delete-inactive-partition-after", 13*time.Hour, "How long to wait before an INACTIVE partition is eligible for deletion. The partition is deleted only if it has been in INACTIVE state for at least the configured duration and it has no owners registered. A value of 0 disables partitions deletion.")
}

func (cfg *PartitionRingConfig) ToLifecyclerConfig(partitionID int32, instanceID string) ring.PartitionInstanceLifecyclerConfig {
func (cfg *Config) ToLifecyclerConfig(partitionID int32, instanceID string) ring.PartitionInstanceLifecyclerConfig {
return ring.PartitionInstanceLifecyclerConfig{
PartitionID: partitionID,
InstanceID: instanceID,
Expand Down
64 changes: 32 additions & 32 deletions pkg/loki/loki.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -85,38 +85,38 @@ type Config struct {
HTTPPrefix string `yaml:"http_prefix" doc:"hidden"`
BallastBytes int `yaml:"ballast_bytes"`

Server server.Config `yaml:"server,omitempty"`
InternalServer internalserver.Config `yaml:"internal_server,omitempty" doc:"hidden"`
Distributor distributor.Config `yaml:"distributor,omitempty"`
Querier querier.Config `yaml:"querier,omitempty"`
QuerierRF1 querierrf1.Config `yaml:"querier_rf1,omitempty"`
QueryScheduler scheduler.Config `yaml:"query_scheduler"`
Frontend lokifrontend.Config `yaml:"frontend,omitempty"`
QueryRange queryrange.Config `yaml:"query_range,omitempty"`
Ruler ruler.Config `yaml:"ruler,omitempty"`
IngesterClient ingester_client.Config `yaml:"ingester_client,omitempty"`
IngesterRF1Client ingester_client.Config `yaml:"ingester_rf1_client,omitempty"`
Ingester ingester.Config `yaml:"ingester,omitempty"`
IngesterRF1 ingester_rf1.Config `yaml:"ingester_rf1,omitempty" category:"experimental"`
PartitionRingConfig partitionring.PartitionRingConfig `yaml:"partition_ring,omitempty" category:"experimental"`
KafkaConfig kafka.Config `yaml:"kafka_ingester,omitempty" category:"experimental"`
Pattern pattern.Config `yaml:"pattern_ingester,omitempty"`
IndexGateway indexgateway.Config `yaml:"index_gateway"`
BloomCompactor bloomcompactor.Config `yaml:"bloom_compactor,omitempty" category:"experimental"`
BloomBuild bloombuild.Config `yaml:"bloom_build,omitempty" category:"experimental"`
BloomGateway bloomgateway.Config `yaml:"bloom_gateway,omitempty" category:"experimental"`
StorageConfig storage.Config `yaml:"storage_config,omitempty"`
ChunkStoreConfig config.ChunkStoreConfig `yaml:"chunk_store_config,omitempty"`
SchemaConfig config.SchemaConfig `yaml:"schema_config,omitempty"`
CompactorConfig compactor.Config `yaml:"compactor,omitempty"`
CompactorHTTPClient compactorclient.HTTPConfig `yaml:"compactor_client,omitempty" doc:"hidden"`
CompactorGRPCClient compactorclient.GRPCConfig `yaml:"compactor_grpc_client,omitempty"`
LimitsConfig validation.Limits `yaml:"limits_config"`
Worker worker.Config `yaml:"frontend_worker,omitempty"`
TableManager index.TableManagerConfig `yaml:"table_manager,omitempty"`
MemberlistKV memberlist.KVConfig `yaml:"memberlist"`
Metastore metastore.Config `yaml:"metastore,omitempty"`
MetastoreClient metastoreclient.Config `yaml:"metastore_client"`
Server server.Config `yaml:"server,omitempty"`
InternalServer internalserver.Config `yaml:"internal_server,omitempty" doc:"hidden"`
Distributor distributor.Config `yaml:"distributor,omitempty"`
Querier querier.Config `yaml:"querier,omitempty"`
QuerierRF1 querierrf1.Config `yaml:"querier_rf1,omitempty"`
QueryScheduler scheduler.Config `yaml:"query_scheduler"`
Frontend lokifrontend.Config `yaml:"frontend,omitempty"`
QueryRange queryrange.Config `yaml:"query_range,omitempty"`
Ruler ruler.Config `yaml:"ruler,omitempty"`
IngesterClient ingester_client.Config `yaml:"ingester_client,omitempty"`
IngesterRF1Client ingester_client.Config `yaml:"ingester_rf1_client,omitempty"`
Ingester ingester.Config `yaml:"ingester,omitempty"`
IngesterRF1 ingester_rf1.Config `yaml:"ingester_rf1,omitempty" category:"experimental"`
PartitionRingConfig partitionring.Config `yaml:"partition_ring,omitempty" category:"experimental"`
KafkaConfig kafka.Config `yaml:"kafka_ingester,omitempty" category:"experimental"`
Pattern pattern.Config `yaml:"pattern_ingester,omitempty"`
IndexGateway indexgateway.Config `yaml:"index_gateway"`
BloomCompactor bloomcompactor.Config `yaml:"bloom_compactor,omitempty" category:"experimental"`
BloomBuild bloombuild.Config `yaml:"bloom_build,omitempty" category:"experimental"`
BloomGateway bloomgateway.Config `yaml:"bloom_gateway,omitempty" category:"experimental"`
StorageConfig storage.Config `yaml:"storage_config,omitempty"`
ChunkStoreConfig config.ChunkStoreConfig `yaml:"chunk_store_config,omitempty"`
SchemaConfig config.SchemaConfig `yaml:"schema_config,omitempty"`
CompactorConfig compactor.Config `yaml:"compactor,omitempty"`
CompactorHTTPClient compactorclient.HTTPConfig `yaml:"compactor_client,omitempty" doc:"hidden"`
CompactorGRPCClient compactorclient.GRPCConfig `yaml:"compactor_grpc_client,omitempty"`
LimitsConfig validation.Limits `yaml:"limits_config"`
Worker worker.Config `yaml:"frontend_worker,omitempty"`
TableManager index.TableManagerConfig `yaml:"table_manager,omitempty"`
MemberlistKV memberlist.KVConfig `yaml:"memberlist"`
Metastore metastore.Config `yaml:"metastore,omitempty"`
MetastoreClient metastoreclient.Config `yaml:"metastore_client"`

RuntimeConfig runtimeconfig.Config `yaml:"runtime_config,omitempty"`
OperationalConfig runtime.Config `yaml:"operational_config,omitempty"`
Expand Down
2 changes: 1 addition & 1 deletion pkg/loki/modules.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -345,7 +345,7 @@ func (t *Loki) initDistributor() (services.Service, error) {
t.Tee = distributor.WrapTee(t.Tee, rf1Tee)
}
if t.Cfg.KafkaConfig.Enabled {
kafkaTee, err := kafka.NewKafkaTee(t.Cfg.KafkaConfig, t.Cfg.MetricsNamespace, prometheus.DefaultRegisterer, util_log.Logger, t.partitionRing)
kafkaTee, err := kafka.NewTee(t.Cfg.KafkaConfig, t.Cfg.MetricsNamespace, prometheus.DefaultRegisterer, util_log.Logger, t.partitionRing)
if err != nil {
return nil, err
}
Expand Down

0 comments on commit 72d8af1

Please sign in to comment.