Gracefully shutdown server (#3896)

cmd/server/grpc_server.go

@@ -0,0 +1,88 @@
+// Copyright 2024 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"net"
+
+	"github.com/rs/zerolog/log"
+	"github.com/urfave/cli/v2"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/keepalive"
+
+	"go.woodpecker-ci.org/woodpecker/v2/pipeline/rpc/proto"
+	"go.woodpecker-ci.org/woodpecker/v2/server"
+	woodpeckerGrpcServer "go.woodpecker-ci.org/woodpecker/v2/server/grpc"
+	"go.woodpecker-ci.org/woodpecker/v2/server/store"
+)
+
+func runGrpcServer(ctx context.Context, c *cli.Context, _store store.Store) error {
+	lis, err := net.Listen("tcp", c.String("grpc-addr"))
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to listen on grpc-addr") //nolint:forbidigo
+	}
+
+	jwtSecret := c.String("grpc-secret")
+	jwtManager := woodpeckerGrpcServer.NewJWTManager(jwtSecret)
+
+	authorizer := woodpeckerGrpcServer.NewAuthorizer(jwtManager)
+	grpcServer := grpc.NewServer(
+		grpc.StreamInterceptor(authorizer.StreamInterceptor),
+		grpc.UnaryInterceptor(authorizer.UnaryInterceptor),
+		grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
+			MinTime: c.Duration("keepalive-min-time"),
+		}),
+	)
+
+	woodpeckerServer := woodpeckerGrpcServer.NewWoodpeckerServer(
+		server.Config.Services.Queue,
+		server.Config.Services.Logs,
+		server.Config.Services.Pubsub,
+		_store,
+	)
+	proto.RegisterWoodpeckerServer(grpcServer, woodpeckerServer)
+
+	woodpeckerAuthServer := woodpeckerGrpcServer.NewWoodpeckerAuthServer(
+		jwtManager,
+		server.Config.Server.AgentToken,
+		_store,
+	)
+	proto.RegisterWoodpeckerAuthServer(grpcServer, woodpeckerAuthServer)
+
+	grpcCtx, cancel := context.WithCancelCause(ctx)
+	defer cancel(nil)
+
+	go func() {
+		<-grpcCtx.Done()
+		if grpcServer == nil {
+			return
+		}
+		log.Info().Msg("terminating grpc service gracefully")
+		grpcServer.GracefulStop()
+		log.Info().Msg("grpc service stopped")
+	}()
+
+	if err := grpcServer.Serve(lis); err != nil {
+		// signal that we don't have to stop the server gracefully anymore
+		grpcServer = nil
+
+		// wrap the error so we know where it did come from
+		return fmt.Errorf("grpc server failed: %w", err)
+	}
+
+	return nil
+}

cmd/server/main.go

@@ -43,6 +43,6 @@ func main() {
 	setupSwaggerStaticConfig()
 
 	if err := app.Run(os.Args); err != nil {
-		log.Fatal().Err(err).Msgf("error running server") //nolint:forbidigo
+		log.Error().Err(err).Msgf("error running server")
 	}
 }

cmd/server/metrics_server.go

@@ -0,0 +1,108 @@
+// Copyright 2024 Woodpecker Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	prometheus_auto "github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/rs/zerolog/log"
+
+	"go.woodpecker-ci.org/woodpecker/v2/server"
+	"go.woodpecker-ci.org/woodpecker/v2/server/store"
+)
+
+func startMetricsCollector(ctx context.Context, _store store.Store) {
+	pendingSteps := prometheus_auto.NewGauge(prometheus.GaugeOpts{
+		Namespace: "woodpecker",
+		Name:      "pending_steps",
+		Help:      "Total number of pending pipeline steps.",
+	})
+	waitingSteps := prometheus_auto.NewGauge(prometheus.GaugeOpts{
+		Namespace: "woodpecker",
+		Name:      "waiting_steps",
+		Help:      "Total number of pipeline waiting on deps.",
+	})
+	runningSteps := prometheus_auto.NewGauge(prometheus.GaugeOpts{
+		Namespace: "woodpecker",
+		Name:      "running_steps",
+		Help:      "Total number of running pipeline steps.",
+	})
+	workers := prometheus_auto.NewGauge(prometheus.GaugeOpts{
+		Namespace: "woodpecker",
+		Name:      "worker_count",
+		Help:      "Total number of workers.",
+	})
+	pipelines := prometheus_auto.NewGauge(prometheus.GaugeOpts{
+		Namespace: "woodpecker",
+		Name:      "pipeline_total_count",
+		Help:      "Total number of pipelines.",
+	})
+	users := prometheus_auto.NewGauge(prometheus.GaugeOpts{
+		Namespace: "woodpecker",
+		Name:      "user_count",
+		Help:      "Total number of users.",
+	})
+	repos := prometheus_auto.NewGauge(prometheus.GaugeOpts{
+		Namespace: "woodpecker",
+		Name:      "repo_count",
+		Help:      "Total number of repos.",
+	})
+
+	go func() {
+		log.Info().Msg("queue metric collector started")
+
+		for {
+			stats := server.Config.Services.Queue.Info(ctx)
+			pendingSteps.Set(float64(stats.Stats.Pending))
+			waitingSteps.Set(float64(stats.Stats.WaitingOnDeps))
+			runningSteps.Set(float64(stats.Stats.Running))
+			workers.Set(float64(stats.Stats.Workers))
+
+			select {
+			case <-ctx.Done():
+				log.Info().Msg("queue metric collector stopped")
+				return
+			case <-time.After(queueInfoRefreshInterval):
+			}
+		}
+	}()
+	go func() {
+		log.Info().Msg("store metric collector started")
+
+		for {
+			repoCount, repoErr := _store.GetRepoCount()
+			userCount, userErr := _store.GetUserCount()
+			pipelineCount, pipelineErr := _store.GetPipelineCount()
+			pipelines.Set(float64(pipelineCount))
+			users.Set(float64(userCount))
+			repos.Set(float64(repoCount))
+
+			if err := errors.Join(repoErr, userErr, pipelineErr); err != nil {
+				log.Error().Err(err).Msg("could not update store information for metrics")
+			}
+
+			select {
+			case <-ctx.Done():
+				log.Info().Msg("store metric collector stopped")
+				return
+			case <-time.After(storeInfoRefreshInterval):
+			}
+		}
+	}()
+}