15584626810

Committed 11 Jun 2025 12:17PM UTC coverage: 64.904% (-0.5%) from 65.4%

Build # 15584626810

Build Type

push

github

Committed by

web-flow

Commit Message

tools: moved to new go 1.24 approach using go get -tool and upgraded to golangci v2 (#1009)

Run Details

26 of 49 new or added lines in 17 files covered. (53.06%)

22 existing lines in 1 file now uncovered.

3645 of 5616 relevant lines covered (64.9%)

3.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/pkg/monitor/server/server.go

package server

import (
        "context"
        "net"
        "os"
        "time"

        log "github.com/sirupsen/logrus"
        "google.golang.org/grpc"
        "google.golang.org/protobuf/types/known/timestamppb"

        "github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/config"
        "github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/gitlab"
        "github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/monitor"
        pb "github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/monitor/protobuf"
        "github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/schemas"
        "github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/store"
)

// Server ..
type Server struct {
        pb.UnimplementedMonitorServer

        gitlabClient             *gitlab.Client
        cfg                      config.Config
        store                    store.Store
        taskSchedulingMonitoring map[schemas.TaskType]*monitor.TaskSchedulingStatus
}

// NewServer ..
func NewServer(
        gitlabClient *gitlab.Client,
        c config.Config,
        st store.Store,
        tsm map[schemas.TaskType]*monitor.TaskSchedulingStatus,
) (s *Server) {
        s = &Server{
                gitlabClient:             gitlabClient,
                cfg:                      c,
                store:                    st,
                taskSchedulingMonitoring: tsm,
        }

        return
}

// Serve ..
func (s *Server) Serve() {
        if s.cfg.Global.InternalMonitoringListenerAddress == nil {
                log.Info("internal monitoring listener address not set")

                return
        }

        log.WithFields(log.Fields{
                "scheme": s.cfg.Global.InternalMonitoringListenerAddress.Scheme,
                "host":   s.cfg.Global.InternalMonitoringListenerAddress.Host,
                "path":   s.cfg.Global.InternalMonitoringListenerAddress.Path,
        }).Info("internal monitoring listener set")

        grpcServer := grpc.NewServer()
        pb.RegisterMonitorServer(grpcServer, s)

        var (
                l   net.Listener
                err error
        )

        switch s.cfg.Global.InternalMonitoringListenerAddress.Scheme {
        case "unix":
                unixAddr, err := net.ResolveUnixAddr("unix", s.cfg.Global.InternalMonitoringListenerAddress.Path)
                if err != nil {
                        log.WithError(err).Fatal()
                }

                if _, err := os.Stat(s.cfg.Global.InternalMonitoringListenerAddress.Path); err == nil {
                        if err := os.Remove(s.cfg.Global.InternalMonitoringListenerAddress.Path); err != nil {
                                log.WithError(err).Fatal()
                        }
                }

                defer func(path string) {
                        if err := os.Remove(path); err != nil {
                                log.WithError(err).Fatal()
                        }
                }(s.cfg.Global.InternalMonitoringListenerAddress.Path)

                if l, err = net.ListenUnix("unix", unixAddr); err != nil {
                        log.WithError(err).Fatal()
                }

        default:
                if l, err = net.Listen(s.cfg.Global.InternalMonitoringListenerAddress.Scheme, s.cfg.Global.InternalMonitoringListenerAddress.Host); err != nil {
                        log.WithError(err).Fatal()
                }
        }

        defer func() { _ = l.Close() }()

        if err = grpcServer.Serve(l); err != nil {
                log.WithError(err).Fatal()
        }
}

// GetConfig ..
func (s *Server) GetConfig(ctx context.Context, _ *pb.Empty) (*pb.Config, error) {
        return &pb.Config{
                Content: s.cfg.ToYAML(),
        }, nil
}

// GetTelemetry ..
func (s *Server) GetTelemetry(_ *pb.Empty, ts pb.Monitor_GetTelemetryServer) (err error) {
        ctx := ts.Context()
        ticker := time.NewTicker(time.Second)

        for {
                telemetry := &pb.Telemetry{
                        Projects: &pb.Entity{},
                        Envs:     &pb.Entity{},
                        Refs:     &pb.Entity{},
                        Metrics:  &pb.Entity{},
                }

                telemetry.GitlabApiUsage = float64(s.gitlabClient.RateCounter.Rate()) / float64(s.cfg.Gitlab.MaximumRequestsPerSecond)
                if telemetry.GitlabApiUsage > 1 {
                        telemetry.GitlabApiUsage = 1
                }

                telemetry.GitlabApiRequestsCount = s.gitlabClient.RequestsCounter.Load()

                telemetry.GitlabApiRateLimit = float64(s.gitlabClient.RequestsRemaining) / float64(s.gitlabClient.RequestsLimit)
                if telemetry.GitlabApiRateLimit > 1 {
                        telemetry.GitlabApiRateLimit = 1
                }

                telemetry.GitlabApiLimitRemaining = uint64(s.gitlabClient.RequestsRemaining)

                var queuedTasks uint64

                queuedTasks, err = s.store.CurrentlyQueuedTasksCount(ctx)
                if err != nil {
                        return
                }

                telemetry.TasksBufferUsage = float64(queuedTasks) / float64(s.cfg.Gitlab.MaximumJobsQueueSize)

                telemetry.TasksExecutedCount, err = s.store.ExecutedTasksCount(ctx)
                if err != nil {
                        return
                }

                telemetry.Projects.Count, err = s.store.ProjectsCount(ctx)
                if err != nil {
                        return
                }

                telemetry.Envs.Count, err = s.store.EnvironmentsCount(ctx)
                if err != nil {
                        return
                }

                telemetry.Refs.Count, err = s.store.RefsCount(ctx)
                if err != nil {
                        return
                }

                telemetry.Metrics.Count, err = s.store.MetricsCount(ctx)
                if err != nil {
                        return
                }

                if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypePullProjectsFromWildcards]; ok {
                        telemetry.Projects.LastPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullProjectsFromWildcards].Last)
                        telemetry.Projects.NextPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullProjectsFromWildcards].Next)
                }

                if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectProjects]; ok {
                        telemetry.Projects.LastGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectProjects].Last)
                        telemetry.Projects.NextGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectProjects].Next)
                }

                if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypePullEnvironmentsFromProjects]; ok {
                        telemetry.Envs.LastPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullEnvironmentsFromProjects].Last)
                        telemetry.Envs.NextPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullEnvironmentsFromProjects].Next)
                }

                if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectEnvironments]; ok {
                        telemetry.Envs.LastGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectEnvironments].Last)
                        telemetry.Envs.NextGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectEnvironments].Next)
                }

                if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypePullRefsFromProjects]; ok {
                        telemetry.Refs.LastPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullRefsFromProjects].Last)
                        telemetry.Refs.NextPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullRefsFromProjects].Next)
                }

                if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectRefs]; ok {
                        telemetry.Refs.LastGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectRefs].Last)
                        telemetry.Refs.NextGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectRefs].Next)
                }

                if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypePullMetrics]; ok {
                        telemetry.Metrics.LastPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullMetrics].Last)
                        telemetry.Metrics.NextPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullMetrics].Next)
                }

                if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectMetrics]; ok {
                        telemetry.Metrics.LastGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectMetrics].Last)
                        telemetry.Metrics.NextGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectMetrics].Next)
                }

                _ = ts.Send(telemetry)

                select {
                case <-ctx.Done():
                        return
                case <-ticker.C:
                        time.Sleep(time.Second)
                }
        }
}

1	package server
2
3	import (
4	"context"
5	"net"
6	"os"
7	"time"
8
9	log "github.com/sirupsen/logrus"
10	"google.golang.org/grpc"
11	"google.golang.org/protobuf/types/known/timestamppb"
12
13	"github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/config"
14	"github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/gitlab"
15	"github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/monitor"
16	pb "github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/monitor/protobuf"
17	"github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/schemas"
18	"github.com/mvisonneau/gitlab-ci-pipelines-exporter/pkg/store"
19	)
20
21	// Server ..
22	type Server struct {
23	pb.UnimplementedMonitorServer
24
25	gitlabClient *gitlab.Client
26	cfg config.Config
27	store store.Store
28	taskSchedulingMonitoring map[schemas.TaskType]*monitor.TaskSchedulingStatus
29	}
30
31	// NewServer ..
32	func NewServer(
33	gitlabClient *gitlab.Client,
34	c config.Config,
35	st store.Store,
36	tsm map[schemas.TaskType]*monitor.TaskSchedulingStatus,
37	) (s *Server) {	×
38	s = &Server{	×
39	gitlabClient: gitlabClient,	×
40	cfg: c,	×
41	store: st,	×
42	taskSchedulingMonitoring: tsm,	×
43	}	×
44		×
45	return	×
46	}	×
47
48	// Serve ..
49	func (s *Server) Serve() {	×
50	if s.cfg.Global.InternalMonitoringListenerAddress == nil {	×
51	log.Info("internal monitoring listener address not set")	×
52		×
53	return	×
54	}	×
55
56	log.WithFields(log.Fields{	×
57	"scheme": s.cfg.Global.InternalMonitoringListenerAddress.Scheme,	×
58	"host": s.cfg.Global.InternalMonitoringListenerAddress.Host,	×
59	"path": s.cfg.Global.InternalMonitoringListenerAddress.Path,	×
60	}).Info("internal monitoring listener set")	×
61		×
62	grpcServer := grpc.NewServer()	×
63	pb.RegisterMonitorServer(grpcServer, s)	×
64		×
65	var (	×
66	l net.Listener	×
67	err error	×
68	)	×
69		×
70	switch s.cfg.Global.InternalMonitoringListenerAddress.Scheme {	×
71	case "unix":	×
72	unixAddr, err := net.ResolveUnixAddr("unix", s.cfg.Global.InternalMonitoringListenerAddress.Path)	×
73	if err != nil {	×
74	log.WithError(err).Fatal()	×
75	}	×
76
77	if _, err := os.Stat(s.cfg.Global.InternalMonitoringListenerAddress.Path); err == nil {	×
78	if err := os.Remove(s.cfg.Global.InternalMonitoringListenerAddress.Path); err != nil {	×
79	log.WithError(err).Fatal()	×
80	}	×
81	}
82
83	defer func(path string) {	×
84	if err := os.Remove(path); err != nil {	×
85	log.WithError(err).Fatal()	×
86	}	×
87	}(s.cfg.Global.InternalMonitoringListenerAddress.Path)
88
89	if l, err = net.ListenUnix("unix", unixAddr); err != nil {	×
90	log.WithError(err).Fatal()	×
91	}	×
92
93	default:	×
94	if l, err = net.Listen(s.cfg.Global.InternalMonitoringListenerAddress.Scheme, s.cfg.Global.InternalMonitoringListenerAddress.Host); err != nil {	×
95	log.WithError(err).Fatal()	×
96	}	×
97	}
98
NEW 99	defer func() { _ = l.Close() }()	×
100
101	if err = grpcServer.Serve(l); err != nil {	×
102	log.WithError(err).Fatal()	×
103	}	×
104	}
105
106	// GetConfig ..
107	func (s Server) GetConfig(ctx context.Context, _ pb.Empty) (*pb.Config, error) {	×
108	return &pb.Config{	×
109	Content: s.cfg.ToYAML(),	×
110	}, nil	×
111	}	×
112
113	// GetTelemetry ..
114	func (s Server) GetTelemetry(_ pb.Empty, ts pb.Monitor_GetTelemetryServer) (err error) {	×
115	ctx := ts.Context()	×
116	ticker := time.NewTicker(time.Second)	×
117		×
118	for {	×
119	telemetry := &pb.Telemetry{	×
120	Projects: &pb.Entity{},	×
121	Envs: &pb.Entity{},	×
122	Refs: &pb.Entity{},	×
123	Metrics: &pb.Entity{},	×
124	}	×
125		×
126	telemetry.GitlabApiUsage = float64(s.gitlabClient.RateCounter.Rate()) / float64(s.cfg.Gitlab.MaximumRequestsPerSecond)	×
127	if telemetry.GitlabApiUsage > 1 {	×
128	telemetry.GitlabApiUsage = 1	×
129	}	×
130
131	telemetry.GitlabApiRequestsCount = s.gitlabClient.RequestsCounter.Load()	×
132		×
133	telemetry.GitlabApiRateLimit = float64(s.gitlabClient.RequestsRemaining) / float64(s.gitlabClient.RequestsLimit)	×
134	if telemetry.GitlabApiRateLimit > 1 {	×
135	telemetry.GitlabApiRateLimit = 1	×
136	}	×
137
138	telemetry.GitlabApiLimitRemaining = uint64(s.gitlabClient.RequestsRemaining)	×
139		×
140	var queuedTasks uint64	×
141		×
142	queuedTasks, err = s.store.CurrentlyQueuedTasksCount(ctx)	×
143	if err != nil {	×
144	return	×
145	}	×
146
147	telemetry.TasksBufferUsage = float64(queuedTasks) / float64(s.cfg.Gitlab.MaximumJobsQueueSize)	×
148		×
149	telemetry.TasksExecutedCount, err = s.store.ExecutedTasksCount(ctx)	×
150	if err != nil {	×
151	return	×
152	}	×
153
154	telemetry.Projects.Count, err = s.store.ProjectsCount(ctx)	×
155	if err != nil {	×
156	return	×
157	}	×
158
159	telemetry.Envs.Count, err = s.store.EnvironmentsCount(ctx)	×
160	if err != nil {	×
161	return	×
162	}	×
163
164	telemetry.Refs.Count, err = s.store.RefsCount(ctx)	×
165	if err != nil {	×
166	return	×
167	}	×
168
169	telemetry.Metrics.Count, err = s.store.MetricsCount(ctx)	×
170	if err != nil {	×
171	return	×
172	}	×
173
174	if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypePullProjectsFromWildcards]; ok {	×
175	telemetry.Projects.LastPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullProjectsFromWildcards].Last)	×
176	telemetry.Projects.NextPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullProjectsFromWildcards].Next)	×
177	}	×
178
179	if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectProjects]; ok {	×
180	telemetry.Projects.LastGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectProjects].Last)	×
181	telemetry.Projects.NextGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectProjects].Next)	×
182	}	×
183
184	if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypePullEnvironmentsFromProjects]; ok {	×
185	telemetry.Envs.LastPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullEnvironmentsFromProjects].Last)	×
186	telemetry.Envs.NextPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullEnvironmentsFromProjects].Next)	×
187	}	×
188
189	if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectEnvironments]; ok {	×
190	telemetry.Envs.LastGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectEnvironments].Last)	×
191	telemetry.Envs.NextGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectEnvironments].Next)	×
192	}	×
193
194	if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypePullRefsFromProjects]; ok {	×
195	telemetry.Refs.LastPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullRefsFromProjects].Last)	×
196	telemetry.Refs.NextPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullRefsFromProjects].Next)	×
197	}	×
198
199	if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectRefs]; ok {	×
200	telemetry.Refs.LastGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectRefs].Last)	×
201	telemetry.Refs.NextGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectRefs].Next)	×
202	}	×
203
204	if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypePullMetrics]; ok {	×
205	telemetry.Metrics.LastPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullMetrics].Last)	×
206	telemetry.Metrics.NextPull = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypePullMetrics].Next)	×
207	}	×
208
209	if _, ok := s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectMetrics]; ok {	×
210	telemetry.Metrics.LastGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectMetrics].Last)	×
211	telemetry.Metrics.NextGc = timestamppb.New(s.taskSchedulingMonitoring[schemas.TaskTypeGarbageCollectMetrics].Next)	×
212	}	×
213
NEW 214	_ = ts.Send(telemetry)	×
215		×
216	select {	×
217	case <-ctx.Done():	×
218	return	×
219	case <-ticker.C:	×
NEW 220	time.Sleep(time.Second)	×
221	}
222	}
223	}

mvisonneau / gitlab-ci-pipelines-exporter / 15584626810

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous