1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
package agent
import (
"context"
"fmt"
"net"
"github.com/ash2k/stager"
"github.com/prometheus/client_golang/prometheus"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/module/modshared"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/module/observability"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/tool/logz"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/internal/tool/prototool"
"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v16/pkg/agentcfg"
"go.uber.org/zap"
)
type module struct {
log *zap.Logger
logLevel zap.AtomicLevel
grpcLogLevel zap.AtomicLevel
defaultGrpcLogLevel agentcfg.LogLevelEnum
api modshared.Api
gatherer prometheus.Gatherer
registerer prometheus.Registerer
listener func() (net.Listener, error)
serverName string
}
const (
prometheusUrlPath = "/metrics"
livenessProbeUrlPath = "/liveness"
readinessProbeUrlPath = "/readiness"
)
func (m *module) Run(ctx context.Context, cfg <-chan *agentcfg.AgentConfiguration) error {
return stager.RunStages(ctx,
func(stage stager.Stage) {
// Listen for config changes and apply to logger
stage.Go(func(ctx context.Context) error {
done := ctx.Done()
for {
select {
case <-done:
return nil
case config, ok := <-cfg:
if !ok {
return nil
}
err := m.setConfigurationLogging(config.Observability.Logging)
if err != nil {
m.log.Error("Failed to apply logging configuration", logz.Error(err))
continue
}
}
}
})
// Start metrics server
stage.Go(func(ctx context.Context) error {
lis, err := m.listener()
if err != nil {
return fmt.Errorf("observability listener failed to start: %w", err)
}
// Error is ignored because metricSrv.Run() closes the listener and
// a second close always produces an error.
defer lis.Close() // nolint:errcheck,gosec
m.log.Info("Observability endpoint is up",
logz.NetNetworkFromAddr(lis.Addr()),
logz.NetAddressFromAddr(lis.Addr()),
)
metricSrv := observability.MetricServer{
Log: m.log,
Api: m.api,
Name: m.serverName,
Listener: lis,
PrometheusUrlPath: prometheusUrlPath,
LivenessProbeUrlPath: livenessProbeUrlPath,
ReadinessProbeUrlPath: readinessProbeUrlPath,
Gatherer: m.gatherer,
Registerer: m.registerer,
ProbeRegistry: observability.NewProbeRegistry(),
}
return metricSrv.Run(ctx)
})
},
)
}
func (m *module) DefaultAndValidateConfiguration(config *agentcfg.AgentConfiguration) error {
prototool.NotNil(&config.Observability)
prototool.NotNil(&config.Observability.Logging)
err := m.defaultAndValidateLogging(config.Observability.Logging)
if err != nil {
return fmt.Errorf("logging: %w", err)
}
return nil
}
func (m *module) Name() string {
return observability.ModuleName
}
func (m *module) defaultAndValidateLogging(logging *agentcfg.LoggingCF) error {
if logging.GrpcLevel == nil {
logging.GrpcLevel = &m.defaultGrpcLogLevel
}
_, err := logz.LevelFromString(logging.Level.String())
if err != nil {
return err
}
_, err = logz.LevelFromString(logging.GrpcLevel.String())
if err != nil {
return err
}
return nil
}
func (m *module) setConfigurationLogging(logging *agentcfg.LoggingCF) error {
err := setLogLevel(m.logLevel, logging.Level)
if err != nil {
return err
}
return setLogLevel(m.grpcLogLevel, *logging.GrpcLevel) // not nil after defaulting
}
func setLogLevel(logLevel zap.AtomicLevel, val agentcfg.LogLevelEnum) error {
level, err := logz.LevelFromString(val.String())
if err != nil {
return err
}
logLevel.SetLevel(level)
return nil
}
|
