85dfe4dcfa
* Begin changes for backend-scheduler module * Comment breadcrumb * Initial compactor extensions and backendscheduler module * Add initial jsonnet * Some cleanup after discussion * Add basic integration test for jobs fetching * Flesh out some unit tests * Clean up and add test coverage * Improve coverage * Clean up dead ring code * Set default value for compactor job polling * Add log message for backend scheduling * Add note about fixing the compactor flow * Small operational fixes * Plumb client defaults * Fix job handling and module dependency chicken/egg/lizard * Update config manifest * Move CompactionBlockSelector * Revert tempodb compactor changes * Revert "Revert tempodb compactor changes" This reverts commit 239c4f2a9df2c289164321f427ac56fa4b7d92cc. * Add backendworker module and work around comapctor dependencies * Move job work and queue into separate package * Add jsonnet for backend-worker * Fix backendscheduler tests * Update docs manifest * Revert compaction method rename * Drop debug * Revert compactor module changes * Drop debug * Revert compactor test change * Fix mock interface after extension * Start e2e integration test * Lint and clean up * Move tmpDir creation to the top * Commit for sharing * Clean up debug * Working integration test * Implement backoff * Add manifest * Clean up comments * Clean up comments * Clean up comments * Clean up * Clean up comments * Clean up test * Clean up comment * Clean up * Ensure the queue is drained before schedule * Fix metric label * Ensure the poller context is canceled * Set default values for worker backoff locally * Fix metric label * Drop worker interval in favor of backoff * Fix prune handling * Improve work selection * Hack in a backend cache * Leverage existing blocklist measurement * Skip the work cache when listing tenants * Test CompactWithConfig * Move proto for new work to new file * Tenants with zero blocks have zero priority * Many things * Add note * Drop empty file * Plumb work config for pruning * Drop needless proto change * Update backendscheduler tests * Update integration tests * Fix config and generate doc manifest * Add config validation * Add more config validation * Implement backoff wrapper for update calls * Move tenant index-writitng to the backend-worker and shard on the ring * Include RingConfig * Generate manifest * Fix manifest instance_id * Drop workerID context injection from the GRPC request * Push job collection into work package * Push logic for skipping blocks which are already working into work package * Log the workerID * Return gRPC error on invalid job status * Metric worker retry calls * Improve test coverage for work * Improve test coverage for job * Fix test logic * Reduce surface area, improve test coverage, clean up error handling * Include new errors * Fix lint * Add tests for backendworker module * Drop comments * Fix backoff for Next() call on the worker to ensure backoff is honored with error return * Bring back outstanding block metric on scheduler * Make note about losing the last work time when the tenant is pushed back to the queue * Measure all tenants, not just ones which get scheduled * Continue scheduling if a single tenant was not enough work to load * thoughts Signed-off-by: Joe Elliott <number101010@gmail.com> * Add a timeout to cleanup jobs which have been running too long * Drop "last work" from tenant selector implementation for simplicity * Revert blocklist changes * Ensure locking on work.OnBlock * Clean up * Generate manifest * Clean up comment * Clean up * Clean up and handle a couple TODO * Metric jobs created * Fix job count per tenant * Fix signature change in grpc DialOption * Generate manifest * Clean up * Spellcheck * Add missing Prune call * Auto start jobs which are returned, metric job retries * Drop Type from NextRequest * Move marshal/unmarshal to Work and drop a mutex * Simplify mutex handling * Test for job reload * Clean up after rebase * Generate manifest --------- Signed-off-by: Joe Elliott <number101010@gmail.com> Co-authored-by: Joe Elliott <number101010@gmail.com>
51 lines
1.6 KiB
Go
51 lines
1.6 KiB
Go
package compactor
|
|
|
|
import (
|
|
"flag"
|
|
"net"
|
|
"strconv"
|
|
|
|
"github.com/go-kit/log"
|
|
"github.com/grafana/dskit/flagext"
|
|
"github.com/grafana/dskit/ring"
|
|
"github.com/grafana/tempo/pkg/util"
|
|
"github.com/grafana/tempo/tempodb"
|
|
)
|
|
|
|
type Config struct {
|
|
Disabled bool `yaml:"disabled,omitempty"`
|
|
ShardingRing RingConfig `yaml:"ring,omitempty"`
|
|
Compactor tempodb.CompactorConfig `yaml:"compaction"`
|
|
OverrideRingKey string `yaml:"override_ring_key"`
|
|
}
|
|
|
|
// RegisterFlagsAndApplyDefaults registers the flags.
|
|
func (cfg *Config) RegisterFlagsAndApplyDefaults(prefix string, f *flag.FlagSet) {
|
|
cfg.Compactor = tempodb.CompactorConfig{}
|
|
cfg.Compactor.RegisterFlagsAndApplyDefaults(util.PrefixConfig(prefix, "compaction"), f)
|
|
|
|
flagext.DefaultValues(&cfg.ShardingRing)
|
|
cfg.ShardingRing.KVStore.Store = "" // by default compactor is not sharded
|
|
|
|
f.BoolVar(&cfg.Disabled, util.PrefixConfig(prefix, "disabled"), false, "Disable compaction.")
|
|
cfg.OverrideRingKey = compactorRingKey
|
|
}
|
|
|
|
func toBasicLifecyclerConfig(cfg RingConfig, logger log.Logger) (ring.BasicLifecyclerConfig, error) {
|
|
instanceAddr, err := ring.GetInstanceAddr(cfg.InstanceAddr, cfg.InstanceInterfaceNames, logger, cfg.EnableInet6)
|
|
if err != nil {
|
|
return ring.BasicLifecyclerConfig{}, err
|
|
}
|
|
|
|
instancePort := ring.GetInstancePort(cfg.InstancePort, cfg.ListenPort)
|
|
|
|
instanceAddrPort := net.JoinHostPort(instanceAddr, strconv.Itoa(instancePort))
|
|
|
|
return ring.BasicLifecyclerConfig{
|
|
ID: cfg.InstanceID,
|
|
Addr: instanceAddrPort,
|
|
HeartbeatPeriod: cfg.HeartbeatPeriod,
|
|
NumTokens: ringNumTokens,
|
|
}, nil
|
|
}
|