forked from grafana/agent
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
service/cluster: migrate pkg/cluster to a Flow-native service (grafan…
…a#4870) * service/http: allow dependant services to register custom handlers This updates the HTTP service to allow downstream services to register custom handlers. This will replace ths custom logic for wiring methods for the HTTP service, which currently requires the clustering service to be explicitly passed to the HTTP service. * service/http: move UI-specific routes to a dedicated UI service. The UI service depends on both the HTTP and clustering services to power the API, which means it should be brought out to a dedicated package to avoid a cyclic dependency. * service/cluster: introduce cluster service The cluster service is a replacement for pkg/cluster that implements the github.com/grafana/agent/service.Service API. It has the following changes compared to the previous implementation: * service/cluster always starts a ckit.Node for consistency around the API. When clustering is disabled, service/cluster never joins any peers, and never exposes its cluster HTTP handlers to peers in the cluster. * Much of the behavior previously provided by pkg/cluster is now deferred to the caller: * The caller is expected to determine their own advertise address, including defaulting to a specific interface. * The caller is expected to provide their own function to discover peers, even if that function returns a static list. * The new implementation, as a Flow service, directly informs components about changes to the cluster state instead of having the Flow controller inform components. While the new implementation puts more work on the caller, its API is simplified. Mutually exclusive mechanisms of the previous API (such as configuring JoinPeers and DiscoverPeers) have been simplified into a single, unified mechanism. While this commit introduces the new service, existing code is not updated the use the new service yet. * flowmode: add utilities to build a cluster.Service implementation The implementation of service/cluster shifts more responsibility to the caller than was previously required with pkg/cluster. This commit adds helper utilities to implement that responsibility to retain the same behavior with the new service. * all: replace usages of pkg/cluster.Clusterer with pkg/cluster.Node This changes all usages of pkg/cluster.Clusterer to use an interface instead. Usage of the interface will permit changing out the implementation, particularly with something from service/cluster. This is a temporary change for the purposes of atomic commits. As such, the Clusterer field is not renamed to the more appropriate name of "Node." * service/cluster: temporarily match pkg/cluster.Node interface This is a temporary commit for service/cluster.Cluster to match the pkg/cluster.Node interface. The purpose of this commit is to atomically swap out the implementation of pkg/cluster with service/cluster without needing to remove the hard-coded usage of the interface yet. * flowmode: Replace usage of pkg/cluster with service/cluster. * flow: use GetServiceData API and remove hard-wired cluster usage * flow: remove remaining references to pkg/cluster * service/cluster: remove unused methods Remove methods which were temporarily added to support an atomic commits of switching pkg/cluster usage with service/cluster. * misc: remove pkg/cluster pkg/cluster is now considered dead code in favor of the flow-specific service/cluster API. * component: remove reference to old ClusteredComponent interface The ClusteredComponent interface has been replaced with the interface definition in cluster.Component.
- Loading branch information
Showing
37 changed files
with
931 additions
and
1,109 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,184 @@ | ||
| package flowmode | ||
|
|
||
| import ( | ||
| "fmt" | ||
| stdlog "log" | ||
| "net" | ||
| "os" | ||
| "strconv" | ||
| "time" | ||
|
|
||
| "github.com/go-kit/log" | ||
| "github.com/grafana/agent/service/cluster" | ||
| "github.com/grafana/ckit/advertise" | ||
| "github.com/hashicorp/go-discover" | ||
| "github.com/hashicorp/go-discover/provider/k8s" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| "go.opentelemetry.io/otel/trace" | ||
| ) | ||
|
|
||
| type clusterOptions struct { | ||
| Log log.Logger | ||
| Metrics prometheus.Registerer | ||
| Tracer trace.TracerProvider | ||
|
|
||
| EnableClustering bool | ||
| NodeName string | ||
| AdvertiseAddress string | ||
| ListenAddress string | ||
| JoinPeers []string | ||
| DiscoverPeers string | ||
| RejoinInterval time.Duration | ||
| } | ||
|
|
||
| func buildClusterService(opts clusterOptions) (*cluster.Service, error) { | ||
| listenPort := findPort(opts.ListenAddress, 80) | ||
|
|
||
| config := cluster.Options{ | ||
| Log: opts.Log, | ||
| Metrics: opts.Metrics, | ||
| Tracer: opts.Tracer, | ||
|
|
||
| EnableClustering: opts.EnableClustering, | ||
| NodeName: opts.NodeName, | ||
| AdvertiseAddress: opts.AdvertiseAddress, | ||
| RejoinInterval: opts.RejoinInterval, | ||
| } | ||
|
|
||
| if config.NodeName == "" { | ||
| hostname, err := os.Hostname() | ||
| if err != nil { | ||
| return nil, fmt.Errorf("generating node name: %w", err) | ||
| } | ||
| config.NodeName = hostname | ||
| } | ||
|
|
||
| if config.AdvertiseAddress == "" { | ||
| // TODO(rfratto): allow advertise interfaces to be configurable. | ||
| addr, err := advertise.FirstAddress(advertise.DefaultInterfaces) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("determining advertise address: %w", err) | ||
| } | ||
| config.AdvertiseAddress = fmt.Sprintf("%s:%d", addr.String(), listenPort) | ||
| } else { | ||
| config.AdvertiseAddress = appendDefaultPort(config.AdvertiseAddress, listenPort) | ||
| } | ||
|
|
||
| switch { | ||
| case len(opts.JoinPeers) > 0 && opts.DiscoverPeers != "": | ||
| return nil, fmt.Errorf("at most one of join peers and discover peers may be set") | ||
|
|
||
| case len(opts.JoinPeers) > 0: | ||
| config.DiscoverPeers = newStaticDiscovery(opts.JoinPeers, listenPort) | ||
|
|
||
| case opts.DiscoverPeers != "": | ||
| discoverFunc, err := newDynamicDiscovery(config.Log, opts.DiscoverPeers, listenPort) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| config.DiscoverPeers = discoverFunc | ||
|
|
||
| default: | ||
| // Here, both JoinPeers and DiscoverPeers are empty. This is desirable when | ||
| // starting a seed node that other nodes connect to, so we don't require | ||
| // one of the fields to be set. | ||
| } | ||
|
|
||
| return cluster.New(config) | ||
| } | ||
|
|
||
| func findPort(addr string, defaultPort int) int { | ||
| _, portStr, err := net.SplitHostPort(addr) | ||
| if err != nil { | ||
| return defaultPort | ||
| } | ||
| port, err := strconv.Atoi(portStr) | ||
| if err != nil { | ||
| return defaultPort | ||
| } | ||
| return port | ||
| } | ||
|
|
||
| func appendDefaultPort(addr string, port int) string { | ||
| _, _, err := net.SplitHostPort(addr) | ||
| if err == nil { | ||
| // No error means there was a port in the string | ||
| return addr | ||
| } | ||
| return fmt.Sprintf("%s:%d", addr, port) | ||
| } | ||
|
|
||
| type discoverFunc func() ([]string, error) | ||
|
|
||
| func newStaticDiscovery(peers []string, defaultPort int) discoverFunc { | ||
| return func() ([]string, error) { | ||
| var addrs []string | ||
|
|
||
| for _, addr := range peers { | ||
| addrs = appendJoinAddr(addrs, addr) | ||
| } | ||
|
|
||
| for i := range addrs { | ||
| // Default to using the same advertise port as the local node. This may | ||
| // break in some cases, so the user should make sure the port numbers | ||
| // align on as many nodes as possible. | ||
| addrs[i] = appendDefaultPort(addrs[i], defaultPort) | ||
| } | ||
|
|
||
| return addrs, nil | ||
| } | ||
| } | ||
|
|
||
| func appendJoinAddr(addrs []string, in string) []string { | ||
| _, _, err := net.SplitHostPort(in) | ||
| if err == nil { | ||
| addrs = append(addrs, in) | ||
| return addrs | ||
| } | ||
|
|
||
| ip := net.ParseIP(in) | ||
| if ip != nil { | ||
| addrs = append(addrs, ip.String()) | ||
| return addrs | ||
| } | ||
|
|
||
| _, srvs, err := net.LookupSRV("", "", in) | ||
| if err == nil { | ||
| for _, srv := range srvs { | ||
| addrs = append(addrs, srv.Target) | ||
| } | ||
| } | ||
|
|
||
| return addrs | ||
| } | ||
|
|
||
| func newDynamicDiscovery(l log.Logger, config string, defaultPort int) (discoverFunc, error) { | ||
| providers := make(map[string]discover.Provider, len(discover.Providers)+1) | ||
| for k, v := range discover.Providers { | ||
| providers[k] = v | ||
| } | ||
|
|
||
| // Custom providers that aren't enabled by default | ||
| providers["k8s"] = &k8s.Provider{} | ||
|
|
||
| discoverer, err := discover.New(discover.WithProviders(providers)) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("bootstrapping peer discovery: %w", err) | ||
| } | ||
|
|
||
| return func() ([]string, error) { | ||
| addrs, err := discoverer.Addrs(config, stdlog.New(log.NewStdlibAdapter(l), "", 0)) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("discovering peers: %w", err) | ||
| } | ||
|
|
||
| for i := range addrs { | ||
| // Default to using the same advertise port as the local node. This may | ||
| // break in some cases, so the user should make sure the port numbers | ||
| // align on as many nodes as possible. | ||
| addrs[i] = appendDefaultPort(addrs[i], defaultPort) | ||
| } | ||
|
|
||
| return addrs, nil | ||
| }, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.