diff --git a/cmd/agent/entrypoint.go b/cmd/agent/entrypoint.go index 3f208e0fa92f..370f457a5d9d 100644 --- a/cmd/agent/entrypoint.go +++ b/cmd/agent/entrypoint.go @@ -11,9 +11,7 @@ import ( "syscall" "github.com/gorilla/mux" - integrations "github.com/grafana/agent/pkg/integrations/v2" "github.com/grafana/agent/pkg/logs" - loki "github.com/grafana/agent/pkg/logs" "github.com/grafana/agent/pkg/metrics" "github.com/grafana/agent/pkg/metrics/instance" "github.com/grafana/agent/pkg/traces" @@ -41,9 +39,9 @@ type Entrypoint struct { srv *server.Server promMetrics *metrics.Agent - lokiLogs *loki.Logs + lokiLogs *logs.Logs tempoTraces *traces.Traces - integrations *integrations.Subsystem + integrations config.Integrations reloadListener net.Listener reloadServer *http.Server @@ -96,7 +94,7 @@ func NewEntrypoint(logger *util.Logger, cfg *config.Config, reloader Reloader) ( if err != nil { return nil, err } - ep.integrations, err = integrations.NewSubsystem(logger, integrationGlobals) + ep.integrations, err = config.NewIntegrations(logger, &cfg.Integrations, integrationGlobals) if err != nil { return nil, err } @@ -109,10 +107,10 @@ func NewEntrypoint(logger *util.Logger, cfg *config.Config, reloader Reloader) ( return ep, nil } -func (ep *Entrypoint) createIntegrationsGlobals(cfg *config.Config) (integrations.Globals, error) { +func (ep *Entrypoint) createIntegrationsGlobals(cfg *config.Config) (config.IntegrationsGlobals, error) { hostname, err := instance.Hostname() if err != nil { - return integrations.Globals{}, fmt.Errorf("getting hostname: %w", err) + return config.IntegrationsGlobals{}, fmt.Errorf("getting hostname: %w", err) } usingTLS := len(cfg.Server.HTTPTLSConfig.TLSCertPath) > 0 && len(cfg.Server.HTTPTLSConfig.TLSKeyPath) > 0 @@ -121,12 +119,12 @@ func (ep *Entrypoint) createIntegrationsGlobals(cfg *config.Config) (integration scheme = "https" } - return integrations.Globals{ + return config.IntegrationsGlobals{ AgentIdentifier: fmt.Sprintf("%s:%d", hostname, cfg.Server.HTTPListenPort), Metrics: ep.promMetrics, Logs: ep.lokiLogs, Tracing: ep.tempoTraces, - SubsystemOpts: cfg.Integrations, + // TODO(rfratto): set SubsystemOptions here when v1 is removed. AgentBaseURL: &url.URL{ Scheme: scheme, Host: fmt.Sprintf("127.0.0.1:%d", cfg.Server.HTTPListenPort), @@ -171,7 +169,7 @@ func (ep *Entrypoint) ApplyConfig(cfg config.Config) error { if err != nil { level.Error(ep.log).Log("msg", "failed to update integrations", "err", err) failed = true - } else if err := ep.integrations.ApplyConfig(integrationGlobals); err != nil { + } else if err := ep.integrations.ApplyConfig(&cfg.Integrations, integrationGlobals); err != nil { level.Error(ep.log).Log("msg", "failed to update integrations", "err", err) failed = true } diff --git a/pkg/config/config.go b/pkg/config/config.go index 4940d7e28139..b98a2d6cfcb6 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -12,7 +12,6 @@ import ( "github.com/weaveworks/common/server" "github.com/drone/envsubst/v2" - "github.com/grafana/agent/pkg/integrations/v2" "github.com/grafana/agent/pkg/logs" "github.com/grafana/agent/pkg/metrics" "github.com/grafana/agent/pkg/traces" @@ -26,16 +25,16 @@ import ( var DefaultConfig = Config{ // All subsystems with a DefaultConfig should be listed here. Metrics: metrics.DefaultConfig, - Integrations: integrations.DefaultSubsystemOptions, + Integrations: DefaultVersionedIntegrations, } // Config contains underlying configurations for the agent type Config struct { - Server server.Config `yaml:"server,omitempty"` - Metrics metrics.Config `yaml:"metrics,omitempty"` - Integrations integrations.SubsystemOptions `yaml:"integrations,omitempty"` - Traces traces.Config `yaml:"traces,omitempty"` - Logs *logs.Config `yaml:"logs,omitempty"` + Server server.Config `yaml:"server,omitempty"` + Metrics metrics.Config `yaml:"metrics,omitempty"` + Integrations VersionedIntegrations `yaml:"integrations,omitempty"` + Traces traces.Config `yaml:"traces,omitempty"` + Logs *logs.Config `yaml:"logs,omitempty"` // We support a secondary server just for the /-/reload endpoint, since // invoking /-/reload against the primary server can cause the server @@ -117,8 +116,8 @@ func (c *Config) ApplyDefaults() error { c.Metrics.ServiceConfig.Lifecycler.ListenPort = c.Server.GRPCListenPort - if len(c.Integrations.PrometheusRemoteWrite) == 0 { - c.Integrations.PrometheusRemoteWrite = c.Metrics.Global.RemoteWrite + if err := c.Integrations.ApplyDefaults(&c.Server, &c.Metrics); err != nil { + return err } // since the Traces config might rely on an existing Loki config @@ -198,14 +197,16 @@ func load(fs *flag.FlagSet, args []string, loader func(string, bool, *Config) er var ( cfg = DefaultConfig - printVersion bool - file string - configExpandEnv bool + printVersion bool + file string + configExpandEnv bool + useIntegrationsV2 bool ) fs.StringVar(&file, "config.file", "", "configuration file to load") fs.BoolVar(&printVersion, "version", false, "Print this build's version information") fs.BoolVar(&configExpandEnv, "config.expand-env", false, "Expands ${var} in config according to the values of the environment variables.") + fs.BoolVar(&useIntegrationsV2, "experiment.integrations-next.enable", false, "Enable next-gen integrations.") cfg.RegisterFlags(fs) if err := fs.Parse(args); err != nil { @@ -229,10 +230,19 @@ func load(fs *flag.FlagSet, args []string, loader func(string, bool, *Config) er return nil, fmt.Errorf("error parsing flags: %w", err) } + // Complete unmarshaling integrations using the version from the flag. This + // MUST be called before ApplyDefaults. + version := integrationsVersion1 + if useIntegrationsV2 { + version = integrationsVersion2 + } + if err := cfg.Integrations.setVersion(version); err != nil { + return nil, fmt.Errorf("error loading config file %s: %w", file, err) + } + // Finally, apply defaults to config that wasn't specified by file or flag if err := cfg.ApplyDefaults(); err != nil { return nil, fmt.Errorf("error in config file: %w", err) } - return &cfg, nil } diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 5be4d3248cb3..e088e61a3c02 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -6,7 +6,6 @@ import ( "testing" "time" - "github.com/grafana/agent/pkg/integrations/v2" "github.com/grafana/agent/pkg/metrics" "github.com/grafana/agent/pkg/metrics/instance" "github.com/grafana/agent/pkg/util" @@ -149,7 +148,7 @@ func TestConfig_Defaults(t *testing.T) { require.NoError(t, err) require.Equal(t, metrics.DefaultConfig, c.Metrics) - require.Equal(t, integrations.DefaultSubsystemOptions, c.Integrations) + require.Equal(t, DefaultVersionedIntegrations, c.Integrations) } func TestConfig_TracesLokiValidates(t *testing.T) { diff --git a/pkg/config/integrations.go b/pkg/config/integrations.go new file mode 100644 index 000000000000..9cbec36cea44 --- /dev/null +++ b/pkg/config/integrations.go @@ -0,0 +1,148 @@ +package config + +import ( + "fmt" + "reflect" + + "github.com/go-kit/log" + "github.com/gorilla/mux" + v1 "github.com/grafana/agent/pkg/integrations" + v2 "github.com/grafana/agent/pkg/integrations/v2" + "github.com/grafana/agent/pkg/metrics" + "github.com/grafana/agent/pkg/util" + "github.com/prometheus/statsd_exporter/pkg/level" + "github.com/weaveworks/common/server" + "gopkg.in/yaml.v2" +) + +type integrationsVersion int + +const ( + integrationsVersion1 integrationsVersion = iota + integrationsVersion2 +) + +// DefaultVersionedIntegrations is the default config for integrations. +var DefaultVersionedIntegrations = VersionedIntegrations{ + version: integrationsVersion1, + configV1: &v1.DefaultManagerConfig, +} + +// VersionedIntegrations abstracts the subsystem configs for integrations v1 +// and v2. VersionedIntegrations can only be unmarshaled as part of Load. +type VersionedIntegrations struct { + version integrationsVersion + raw util.RawYAML + + configV1 *v1.ManagerConfig + configV2 *v2.SubsystemOptions +} + +var ( + _ yaml.Unmarshaler = (*VersionedIntegrations)(nil) + _ yaml.Marshaler = (*VersionedIntegrations)(nil) +) + +// UnmarshalYAML implements yaml.Unmarshaler. Full unmarshaling is deferred until +// setVersion is invoked. +func (c *VersionedIntegrations) UnmarshalYAML(unmarshal func(interface{}) error) error { + c.configV1 = nil + c.configV2 = nil + return unmarshal(&c.raw) +} + +// MarshalYAML implements yaml.Marshaler. +func (c VersionedIntegrations) MarshalYAML() (interface{}, error) { + switch { + case c.configV1 != nil: + return c.configV1, nil + case c.configV2 != nil: + return c.configV2, nil + default: + return c.raw, nil + } +} + +// IsZero implements yaml.IsZeroer. +func (c VersionedIntegrations) IsZero() bool { + switch { + case c.configV1 != nil: + return reflect.ValueOf(*c.configV1).IsZero() + case c.configV2 != nil: + return reflect.ValueOf(*c.configV2).IsZero() + default: + return len(c.raw) == 0 + } +} + +// ApplyDefaults applies defaults to the subsystem based on globals. +func (c *VersionedIntegrations) ApplyDefaults(scfg *server.Config, mcfg *metrics.Config) error { + if c.version != integrationsVersion2 { + return c.configV1.ApplyDefaults(scfg, mcfg) + } + return c.configV2.ApplyDefaults(mcfg) +} + +// setVersion completes the deferred unmarshal and unmarshals the raw YAML into +// the subsystem config for version v. +func (c *VersionedIntegrations) setVersion(v integrationsVersion) error { + c.version = v + + switch c.version { + case integrationsVersion1: + cfg := v1.DefaultManagerConfig + c.configV1 = &cfg + return yaml.UnmarshalStrict(c.raw, c.configV1) + case integrationsVersion2: + cfg := v2.DefaultSubsystemOptions + c.configV2 = &cfg + return yaml.UnmarshalStrict(c.raw, c.configV2) + default: + panic(fmt.Sprintf("unknown integrations version %d", c.version)) + } +} + +// IntegrationsGlobals is a global struct shared across integrations. +type IntegrationsGlobals = v2.Globals + +// Integrations is an abstraction over both the v1 and v2 systems. +type Integrations interface { + ApplyConfig(*VersionedIntegrations, IntegrationsGlobals) error + WireAPI(*mux.Router) + Stop() +} + +// NewIntegrations creates a new subsystem. globals should be provided regardless +// of useV2. globals.SubsystemOptions will be automatically set if cfg.Version +// is set to IntegrationsVersion2. +func NewIntegrations(logger log.Logger, cfg *VersionedIntegrations, globals IntegrationsGlobals) (Integrations, error) { + if cfg.version != integrationsVersion2 { + instance, err := v1.NewManager(*cfg.configV1, logger, globals.Metrics.InstanceManager(), globals.Metrics.Validate) + if err != nil { + return nil, err + } + return &v1Integrations{Manager: instance}, nil + } + + level.Warn(logger).Log("msg", "integrations-next is enabled. integrations-next is subject to change") + + globals.SubsystemOpts = *cfg.configV2 + instance, err := v2.NewSubsystem(logger, globals) + if err != nil { + return nil, err + } + return &v2Integrations{Subsystem: instance}, nil +} + +type v1Integrations struct{ *v1.Manager } + +func (s *v1Integrations) ApplyConfig(cfg *VersionedIntegrations, globals IntegrationsGlobals) error { + return s.Manager.ApplyConfig(*cfg.configV1) +} + +type v2Integrations struct{ *v2.Subsystem } + +func (s *v2Integrations) ApplyConfig(cfg *VersionedIntegrations, globals IntegrationsGlobals) error { + globals.SubsystemOpts = *cfg.configV2 + return s.Subsystem.ApplyConfig(globals) +} diff --git a/pkg/integrations/manager.go b/pkg/integrations/manager.go index 0306d58ea7a5..9ceed71eaf19 100644 --- a/pkg/integrations/manager.go +++ b/pkg/integrations/manager.go @@ -24,6 +24,7 @@ import ( promConfig "github.com/prometheus/prometheus/config" "github.com/prometheus/prometheus/discovery" "github.com/prometheus/prometheus/pkg/relabel" + "github.com/weaveworks/common/server" ) var ( @@ -115,7 +116,17 @@ func (c *ManagerConfig) DefaultRelabelConfigs(instanceKey string) []*relabel.Con // // If any integrations are enabled and are configured to be scraped, the // Prometheus configuration must have a WAL directory configured. -func (c *ManagerConfig) ApplyDefaults(cfg *metrics.Config) error { +func (c *ManagerConfig) ApplyDefaults(scfg *server.Config, mcfg *metrics.Config) error { + c.ListenPort = scfg.HTTPListenPort + c.ListenHost = scfg.HTTPListenAddress + + c.ServerUsingTLS = scfg.HTTPTLSConfig.TLSKeyPath != "" && scfg.HTTPTLSConfig.TLSCertPath != "" + + if len(c.PrometheusRemoteWrite) == 0 { + c.PrometheusRemoteWrite = mcfg.Global.RemoteWrite + } + c.PrometheusGlobalConfig = mcfg.Global.Prometheus + for _, ic := range c.Integrations { if !ic.CommonConfig().Enabled { continue @@ -127,7 +138,7 @@ func (c *ManagerConfig) ApplyDefaults(cfg *metrics.Config) error { } // WAL must be configured if an integration is going to be scraped. - if scrapeIntegration && cfg.WALDir == "" { + if scrapeIntegration && mcfg.WALDir == "" { return fmt.Errorf("no wal_directory configured") } } diff --git a/pkg/integrations/v2/subsystem.go b/pkg/integrations/v2/subsystem.go index 28bcab0ac839..319676917c7f 100644 --- a/pkg/integrations/v2/subsystem.go +++ b/pkg/integrations/v2/subsystem.go @@ -10,6 +10,7 @@ import ( "github.com/go-kit/log" "github.com/gorilla/mux" + "github.com/grafana/agent/pkg/metrics" "github.com/grafana/agent/pkg/metrics/instance" common_config "github.com/prometheus/common/config" "github.com/prometheus/common/model" @@ -46,6 +47,15 @@ type SubsystemOptions struct { ClientConfig common_config.HTTPClientConfig `yaml:"client_config,omitempty"` } +// ApplyDefaults will apply defaults to o. +func (o *SubsystemOptions) ApplyDefaults(mcfg *metrics.Config) error { + if len(o.PrometheusRemoteWrite) == 0 { + o.PrometheusRemoteWrite = mcfg.Global.RemoteWrite + } + + return nil +} + // MarshalYAML implements yaml.Marshaler for SubsystemOptions. Integrations // will be marshaled inline. func (o SubsystemOptions) MarshalYAML() (interface{}, error) { diff --git a/pkg/util/rawyaml.go b/pkg/util/rawyaml.go new file mode 100644 index 000000000000..42ef2accaa73 --- /dev/null +++ b/pkg/util/rawyaml.go @@ -0,0 +1,25 @@ +package util + +import "gopkg.in/yaml.v2" + +// RawYAML is similar to json.RawMessage and allows for deferred YAML decoding. +type RawYAML []byte + +// UnmarshalYAML implements yaml.Unmarshaler. +func (r *RawYAML) UnmarshalYAML(unmarshal func(interface{}) error) error { + var ms yaml.MapSlice + if err := unmarshal(&ms); err != nil { + return err + } + bb, err := yaml.Marshal(ms) + if err != nil { + return err + } + *r = bb + return nil +} + +// MarshalYAML implements yaml.Marshaler. +func (r RawYAML) MarshalYAML() (interface{}, error) { + return string(r), nil +}