mirror of
https://github.com/hanzoai/traces.git
synced 2026-10-30 18:20:41 +00:00
b5930aa883
* 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 <[email protected]> * 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 <[email protected]> Co-authored-by: Joe Elliott <[email protected]>
60 lines
1.4 KiB
Go
60 lines
1.4 KiB
Go
package client
|
|
|
|
import (
|
|
"flag"
|
|
"fmt"
|
|
"io"
|
|
|
|
"github.com/grafana/dskit/grpcclient"
|
|
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials/insecure"
|
|
"google.golang.org/grpc/health/grpc_health_v1"
|
|
|
|
"github.com/grafana/tempo/pkg/tempopb"
|
|
)
|
|
|
|
// Config for a backendscheduler client.
|
|
type Config struct {
|
|
GRPCClientConfig grpcclient.Config `yaml:"grpc_client_config"`
|
|
}
|
|
|
|
type Client struct {
|
|
tempopb.BackendSchedulerClient
|
|
grpc_health_v1.HealthClient
|
|
io.Closer
|
|
}
|
|
|
|
// RegisterFlags registers flags.
|
|
func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
|
|
cfg.GRPCClientConfig.RegisterFlagsWithPrefix("backendscheduler.client", f)
|
|
}
|
|
|
|
// New returns a new backendscheduler client.
|
|
func New(addr string, cfg Config) (*Client, error) {
|
|
if addr == "" {
|
|
return nil, fmt.Errorf("backend scheduler address is required")
|
|
}
|
|
|
|
opts := []grpc.DialOption{
|
|
grpc.WithTransportCredentials(insecure.NewCredentials()),
|
|
grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
|
|
}
|
|
|
|
instrumentationOpts, err := cfg.GRPCClientConfig.DialOption(nil, nil, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
opts = append(opts, instrumentationOpts...)
|
|
conn, err := grpc.NewClient(addr, opts...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &Client{
|
|
BackendSchedulerClient: tempopb.NewBackendSchedulerClient(conn),
|
|
HealthClient: grpc_health_v1.NewHealthClient(conn),
|
|
Closer: conn,
|
|
}, nil
|
|
}
|