Skip to content
This repository was archived by the owner on Jul 28, 2026. It is now read-only.
18 changes: 8 additions & 10 deletions cmd/agent/entrypoint.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -41,9 +39,9 @@ type Entrypoint struct {

srv *server.Server
promMetrics *metrics.Agent
lokiLogs *loki.Logs
lokiLogs *logs.Logs

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My editor was complaining about this; we had the package imported twice with a different alias.

tempoTraces *traces.Traces
integrations *integrations.Subsystem
integrations config.Integrations

reloadListener net.Listener
reloadServer *http.Server
Expand Down Expand Up @@ -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
}
Expand All @@ -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
Expand All @@ -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),
Expand Down Expand Up @@ -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
}
Expand Down
36 changes: 23 additions & 13 deletions pkg/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand All @@ -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
}
3 changes: 1 addition & 2 deletions pkg/config/config_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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) {
Expand Down
148 changes: 148 additions & 0 deletions pkg/config/integrations.go
Original file line number Diff line number Diff line change
@@ -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)
}
15 changes: 13 additions & 2 deletions pkg/integrations/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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
Expand All @@ -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")
}
}
Expand Down
10 changes: 10 additions & 0 deletions pkg/integrations/v2/subsystem.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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) {
Expand Down
Loading