diff --git a/agent/rpc/dial.go b/agent/rpc/dial.go new file mode 100644 index 000000000..04f8d2e13 --- /dev/null +++ b/agent/rpc/dial.go @@ -0,0 +1,112 @@ +// Copyright 2026 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 rpc + +import ( + "context" + "crypto/tls" + "fmt" + "time" + + "google.golang.org/grpc" + grpc_credentials "google.golang.org/grpc/credentials" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/keepalive" +) + +// DialConfig bundles everything Dial needs. Callers build this from a +// *cli.Command (production) or with literals (tests). +type DialConfig struct { + ServerAddr string + AgentToken string + AgentID int64 + Secure bool + SkipTLSVerify bool + KeepaliveTime time.Duration + KeepaliveTimeout time.Duration + AuthRefreshEvery time.Duration +} + +// AgentConn holds the two gRPC connections and the auth interceptor an agent +// needs. Callers are responsible for closing both connections and canceling +// the auth context passed to Dial when the agent shuts down. +type AgentConn struct { + AuthConn *grpc.ClientConn + MainConn *grpc.ClientConn + AuthInterceptor *AuthInterceptor + AgentID int64 +} + +// Close closes both connections. Safe to call even if one or both are nil. +func (c *AgentConn) Close() { + if c.MainConn != nil { + _ = c.MainConn.Close() + } + if c.AuthConn != nil { + _ = c.AuthConn.Close() + } +} + +// Dial builds the auth gRPC connection, authenticates, then builds the +// authenticated main gRPC connection. +// +// The authCtx parameter governs the lifetime of the token-refresh goroutine +// inside the interceptor; callers typically want a context separate from +// request ctx so the interceptor survives long-running polls. +func Dial(authCtx context.Context, cfg DialConfig) (*AgentConn, error) { + var transport grpc.DialOption + if cfg.Secure { + transport = grpc.WithTransportCredentials(grpc_credentials.NewTLS( + &tls.Config{InsecureSkipVerify: cfg.SkipTLSVerify}, //nolint:gosec // user-opt-in via DialConfig.SkipTLSVerify + )) + } else { + transport = grpc.WithTransportCredentials(insecure.NewCredentials()) + } + + keepaliveOpts := grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: cfg.KeepaliveTime, + Timeout: cfg.KeepaliveTimeout, + }) + + authConn, err := grpc.NewClient(cfg.ServerAddr, transport, keepaliveOpts) + if err != nil { + return nil, fmt.Errorf("create auth gRPC connection: %w", err) + } + + authClient := NewAuthGrpcClient(authConn, cfg.AgentToken, cfg.AgentID) + reportedAgentID := authClient.AgentID() + authInterceptor, err := NewAuthInterceptor(authCtx, authClient, cfg.AuthRefreshEvery) + if err != nil { + _ = authConn.Close() + return nil, fmt.Errorf("authenticate with server: %w", err) + } + + mainConn, err := grpc.NewClient( + cfg.ServerAddr, transport, keepaliveOpts, + grpc.WithUnaryInterceptor(authInterceptor.Unary()), + grpc.WithStreamInterceptor(authInterceptor.Stream()), + ) + if err != nil { + _ = authConn.Close() + return nil, fmt.Errorf("create main gRPC connection: %w", err) + } + + return &AgentConn{ + AuthConn: authConn, + MainConn: mainConn, + AuthInterceptor: authInterceptor, + AgentID: reportedAgentID, + }, nil +} diff --git a/cmd/agent/core/agent.go b/cmd/agent/core/agent.go index b5b87c6d6..92abe2719 100644 --- a/cmd/agent/core/agent.go +++ b/cmd/agent/core/agent.go @@ -17,7 +17,6 @@ package core import ( "context" - "crypto/tls" "errors" "fmt" "maps" @@ -30,11 +29,7 @@ import ( "github.com/rs/zerolog/log" "github.com/urfave/cli/v3" "golang.org/x/sync/errgroup" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - grpc_credentials "google.golang.org/grpc/credentials" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/keepalive" "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" @@ -106,64 +101,41 @@ func run(ctx context.Context, c *cli.Command, backends []types.Backend) error { ) } - var transport grpc.DialOption - if c.Bool("grpc-secure") { - log.Trace().Msg("use ssl for grpc") - transport = grpc.WithTransportCredentials(grpc_credentials.NewTLS(&tls.Config{InsecureSkipVerify: c.Bool("grpc-skip-insecure")})) - } else { - transport = grpc.WithTransportCredentials(insecure.NewCredentials()) - } - - authConn, err := grpc.NewClient( - c.String("server"), - transport, - grpc.WithKeepaliveParams(keepalive.ClientParameters{ - Time: c.Duration("grpc-keepalive-time"), - Timeout: c.Duration("grpc-keepalive-timeout"), - }), - ) - if err != nil { - return fmt.Errorf("could not create new gRPC 'channel' for authentication: %w", err) - } - defer authConn.Close() - agentConfig := readAgentConfig(agentConfigPath) - agentToken := c.String("grpc-token") grpcClientCtx, grpcClientCtxCancel := context.WithCancelCause(context.Background()) defer grpcClientCtxCancel(nil) - authClient := agent_rpc.NewAuthGrpcClient(authConn, agentToken, agentConfig.AgentID) - authInterceptor, err := agent_rpc.NewAuthInterceptor(grpcClientCtx, authClient, authInterceptorRefreshInterval) //nolint:contextcheck - if err != nil { - return fmt.Errorf("agent could not auth: %w", err) + + if c.Bool("grpc-secure") { + log.Trace().Msg("use ssl for grpc") } + agentConn, err := agent_rpc.Dial(grpcClientCtx, agent_rpc.DialConfig{ //nolint:contextcheck + ServerAddr: c.String("server"), + AgentToken: c.String("grpc-token"), + AgentID: agentConfig.AgentID, + Secure: c.Bool("grpc-secure"), + SkipTLSVerify: c.Bool("grpc-skip-insecure"), + KeepaliveTime: c.Duration("grpc-keepalive-time"), + KeepaliveTimeout: c.Duration("grpc-keepalive-timeout"), + AuthRefreshEvery: authInterceptorRefreshInterval, + }) + if err != nil { + return err + } + defer agentConn.Close() + // Persist the agent ID received during auth so that crashloops reuse the // same server-side entry instead of creating a new one on every restart. if agentConfigPath != "" { - agentConfig.AgentID = authClient.AgentID() + agentConfig.AgentID = agentConn.AgentID if err := writeAgentConfig(agentConfig, agentConfigPath); err == nil { log.Debug().Msgf("persisted agent ID %d after auth", agentConfig.AgentID) } } - conn, err := grpc.NewClient( - c.String("server"), - transport, - grpc.WithKeepaliveParams(keepalive.ClientParameters{ - Time: c.Duration("grpc-keepalive-time"), - Timeout: c.Duration("grpc-keepalive-timeout"), - }), - grpc.WithUnaryInterceptor(authInterceptor.Unary()), - grpc.WithStreamInterceptor(authInterceptor.Stream()), - ) - if err != nil { - return fmt.Errorf("could not create new gRPC 'channel' for normal orchestration: %w", err) - } - defer conn.Close() - client := agent_rpc.NewGrpcClient( - ctx, conn, + ctx, agentConn.MainConn, agent_rpc.SetConnectionRetryTimeout(c.Duration("retry-timeout")), ) agentConfigPersisted := atomic.Bool{} diff --git a/cmd/server/grpc_server.go b/cmd/server/grpc_server.go index a5feae8a0..2430e4b79 100644 --- a/cmd/server/grpc_server.go +++ b/cmd/server/grpc_server.go @@ -19,12 +19,9 @@ import ( "fmt" "net" - "github.com/rs/zerolog/log" + "github.com/prometheus/client_golang/prometheus" "github.com/urfave/cli/v3" - "google.golang.org/grpc" - "google.golang.org/grpc/keepalive" - "go.woodpecker-ci.org/woodpecker/v3/rpc/proto" "go.woodpecker-ci.org/woodpecker/v3/server" server_rpc "go.woodpecker-ci.org/woodpecker/v3/server/rpc" "go.woodpecker-ci.org/woodpecker/v3/server/store" @@ -36,52 +33,14 @@ func runGrpcServer(ctx context.Context, c *cli.Command, _store store.Store) erro return fmt.Errorf("failed to listen on grpc-addr: %w", err) } - jwtSecret := c.String("grpc-secret") - jwtManager := server_rpc.NewJWTManager(jwtSecret) - - authorizer := server_rpc.NewAuthorizer(jwtManager) - grpcServer := grpc.NewServer( - grpc.StreamInterceptor(authorizer.StreamInterceptor), - grpc.UnaryInterceptor(authorizer.UnaryInterceptor), - grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ - MinTime: c.Duration("keepalive-min-time"), - }), - ) - - woodpeckerServer := server_rpc.NewWoodpeckerServer( - server.Config.Services.Scheduler, - server.Config.Services.Logs, - _store, - ) - proto.RegisterWoodpeckerServer(grpcServer, woodpeckerServer) - - woodpeckerAuthServer := server_rpc.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 + return server_rpc.Serve(ctx, server_rpc.ServeConfig{ + Listener: lis, + Store: _store, + Scheduler: server.Config.Services.Scheduler, + Logger: server.Config.Services.Logs, + JWTSecret: c.String("grpc-secret"), + AgentToken: server.Config.Server.AgentToken, + KeepaliveMinTime: c.Duration("keepalive-min-time"), + Registerer: prometheus.DefaultRegisterer, + }) } diff --git a/e2e/scenarios/fixtures.go b/e2e/scenarios/fixtures.go index be965b97a..e05f3d599 100644 --- a/e2e/scenarios/fixtures.go +++ b/e2e/scenarios/fixtures.go @@ -137,9 +137,8 @@ func LoadScenarios(t *testing.T) []Scenario { {Name: ".woodpecker.yaml", Data: yamlData}, } - if s.Event == "" { - s.Event = model.EventPush - } + s.check(t, stem+".json") + scenarios = append(scenarios, s) } @@ -178,9 +177,9 @@ func loadMultiWorkflowScenario(t *testing.T, dirName string) Scenario { }) } } - require.NotEmpty(t, files, "no YAML files in multi-workflow dir %s", dirName) - require.NotEmpty(t, s.Name, "scenario.json missing 'name' in %s", dirName) + + s.check(t, dirName+"/scenario.json") s.Files = files if s.Event == "" { @@ -188,3 +187,14 @@ func loadMultiWorkflowScenario(t *testing.T, dirName string) Scenario { } return s } + +func (s Scenario) check(t *testing.T, location string) { + t.Helper() + + if s.Name == "" { + t.Fatalf("fixture '%s' missing name", location) + } + if s.Event == "" { + t.Fatalf("fixture '%s' missing event", location) + } +} diff --git a/e2e/setup/agent.go b/e2e/setup/agent.go index 6fd4425d5..366b88171 100644 --- a/e2e/setup/agent.go +++ b/e2e/setup/agent.go @@ -23,9 +23,6 @@ import ( "github.com/rs/zerolog/log" "github.com/stretchr/testify/require" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/keepalive" "google.golang.org/grpc/metadata" "go.woodpecker-ci.org/woodpecker/v3/agent" @@ -51,9 +48,8 @@ type AgentEnv struct { // name is used for logging and as the hostname label. name string - // requestedOrgID is applied to the DB record by WaitForAgentRegistered - // so the server's GetServerLabels returns the right org-id filter. - // model.IDNotSet (-1) means global (default). + // requestOrgID is applied to the DB record by WaitForAgentRegistered. + // See WithOrgID. model.IDNotSet (-1) means global (default). requestOrgID int64 } @@ -68,9 +64,7 @@ type agentConfig struct { // They are matched against task Labels set in pipeline YAML (labels: key: value). customLabels map[string]string - // orgID pins the agent to a specific organization (-1 = global). - // Org agents score higher than global agents for tasks in the same org, - // so they are always preferred by the queue when available. + // orgID pins the agent to a specific organization. See WithOrgID. orgID int64 } @@ -82,7 +76,7 @@ func WithHostname(name string) AgentOption { // WithCustomLabels merges extra labels into the agent's filter set. // Use this to test label-based task routing, e.g.: // -// setup.StartAgent(ctx, t, addr, setup.WithCustomLabels(map[string]string{"gpu": "true"})) +// setup.StartAgent(t, addr, setup.WithCustomLabels(map[string]string{"gpu": "true"})) // // The pipeline YAML must set a matching label: // @@ -108,6 +102,11 @@ func WithOrgID(id int64) AgentOption { // server at grpcAddr and returns an *AgentEnv whose AgentID is populated once // the agent has registered. Pass AgentOption values to configure labels, hostname, // or org-scoping; multiple agents can be started in the same test. +// +// The agent owns its own lifetime via an internal context canceled through +// t.Cleanup. This decouples agent shutdown from the test's own context: the +// agent keeps running until the test completes, regardless of which ctx the +// caller's pipeline operations use. func StartAgent(t *testing.T, grpcAddr string, opts ...AgentOption) *AgentEnv { t.Helper() @@ -122,40 +121,24 @@ func StartAgent(t *testing.T, grpcAddr string, opts ...AgentOption) *AgentEnv { env := &AgentEnv{name: cfg.hostname} - transport := grpc.WithTransportCredentials(insecure.NewCredentials()) - keepaliveOpts := grpc.WithKeepaliveParams(keepalive.ClientParameters{ - Time: defaultTimeout, - Timeout: shortTimeout, - }) - agentCtx, agentCancel := context.WithCancelCause(t.Context()) t.Cleanup(func() { agentCancel(nil) }) - authConn, err := grpc.NewClient(grpcAddr, transport, keepaliveOpts) + agentConn, err := agent_rpc.Dial(agentCtx, agent_rpc.DialConfig{ + ServerAddr: grpcAddr, + AgentToken: TestAgentToken, + AgentID: -1, + Secure: false, + KeepaliveTime: defaultTimeout, + KeepaliveTimeout: shortTimeout, + AuthRefreshEvery: agentAuthRefreshEvery, + }) if err != nil { - t.Fatalf("StartAgent(%s): create auth gRPC connection: %v", cfg.hostname, err) + t.Fatalf("StartAgent(%s): dial gRPC: %v", cfg.hostname, err) } - t.Cleanup(func() { authConn.Close() }) + t.Cleanup(agentConn.Close) - authClient := agent_rpc.NewAuthGrpcClient(authConn, TestAgentToken, -1) - authInterceptor, err := agent_rpc.NewAuthInterceptor(agentCtx, authClient, agentAuthRefreshEvery) - if err != nil { - t.Fatalf("StartAgent(%s): authenticate with server: %v", cfg.hostname, err) - } - - conn, err := grpc.NewClient( - grpcAddr, - transport, - keepaliveOpts, - grpc.WithUnaryInterceptor(authInterceptor.Unary()), - grpc.WithStreamInterceptor(authInterceptor.Stream()), - ) - if err != nil { - t.Fatalf("StartAgent(%s): create main gRPC connection: %v", cfg.hostname, err) - } - t.Cleanup(func() { conn.Close() }) - - client := agent_rpc.NewGrpcClient(agentCtx, conn) + client := agent_rpc.NewGrpcClient(agentCtx, agentConn.MainConn) grpcCtx := metadata.NewOutgoingContext(agentCtx, metadata.Pairs("hostname", cfg.hostname)) diff --git a/e2e/setup/forge.go b/e2e/setup/forge.go index d8a9a81f6..948a8abb4 100644 --- a/e2e/setup/forge.go +++ b/e2e/setup/forge.go @@ -17,7 +17,6 @@ package setup import ( - "net/http" "testing" "github.com/stretchr/testify/mock" @@ -29,13 +28,6 @@ import ( // newMockForge builds a MockForge that serves the given files for any // config-fetch call, no-ops status reporting, and stubs all other methods safely. -// -// Single-workflow (len(files)==1, name ".woodpecker.yaml"): File() returns the -// raw YAML bytes; Dir() is not called but is stubbed for safety. -// -// Multi-workflow (len(files)>1, names ".woodpecker/foo.yaml"): File() returns -// empty (causing the config service to fall through to Dir()); Dir() returns -// all files. func newMockForge(t *testing.T, files []*forge_types.FileMeta) *forge_mocks.MockForge { t.Helper() m := forge_mocks.NewMockForge(t) @@ -44,29 +36,16 @@ func newMockForge(t *testing.T, files []*forge_types.FileMeta) *forge_mocks.Mock m.On("Name").Return("mock").Maybe() m.On("URL").Return("https://forge.example.test").Maybe() - if len(files) == 1 { - // Single-workflow: config service calls File(".woodpecker.yaml"). - m.On( - "File", - mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker.yaml", - ).Return(files[0].Data, nil).Maybe() + // we just use multi workflows + m.On( + "File", + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, + ).Return(nil, nil).Maybe() - m.On( - "Dir", - mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker", - ).Return(files, nil).Maybe() - } else { - // Multi-workflow: config service calls Dir(".woodpecker"). - // File() must return empty so the service falls through to Dir(). - m.On( - "File", - mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker.yaml", - ).Return([]byte(nil), nil).Maybe() - m.On( - "Dir", - mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker", - ).Return(files, nil).Maybe() - } + m.On( + "Dir", + mock.Anything, mock.Anything, mock.Anything, mock.Anything, ".woodpecker", + ).Return(files, nil).Maybe() // Status reporting back to forge — no-op. m.On( @@ -82,6 +61,3 @@ func newMockForge(t *testing.T, files []*forge_types.FileMeta) *forge_mocks.Mock return m } - -// compile-time import guard. -var _ *http.Request diff --git a/e2e/setup/server.go b/e2e/setup/server.go index 1e1feeca5..2ab855a33 100644 --- a/e2e/setup/server.go +++ b/e2e/setup/server.go @@ -25,10 +25,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/require" "github.com/urfave/cli/v3" - "google.golang.org/grpc" - "google.golang.org/grpc/keepalive" - "go.woodpecker-ci.org/woodpecker/v3/rpc/proto" "go.woodpecker-ci.org/woodpecker/v3/server" "go.woodpecker-ci.org/woodpecker/v3/server/cache" "go.woodpecker-ci.org/woodpecker/v3/server/forge" @@ -54,7 +51,7 @@ const ( // TestJWTSecret is used for signing gRPC auth JWTs. TestJWTSecret = "test-jwt-secret-for-integration-tests" - // TestForgeType is the forge type the mock pretends to bee. + // TestForgeType is the forge type the mock pretends to be. TestForgeType = model.ForgeTypeGitea ) @@ -149,7 +146,10 @@ func newTestManager(s store.Store, mockForge *forge_mocks.MockForge) (services.M &cli.DurationFlag{Name: "forge-timeout", Value: defaultTimeout}, &cli.UintFlag{Name: "forge-retry", Value: defaultRetry}, &cli.StringSliceFlag{Name: "environment"}, - // Forge flags — gitea=true satisfies setupForgeService's type switch. + // services.NewManager reads the forge type from a cli flag and + // drives a type-switch in setupForgeService. We set the gitea flag + // to satisfy that switch; the actual forge instance is overridden + // below via the SetupForge hook, so the switch result is unused. &cli.BoolFlag{Name: string(TestForgeType), Value: true}, &cli.StringFlag{Name: "forge-url", Value: "https://forge.example.test"}, &cli.StringSliceFlag{Name: "default-pipeline-configs", Value: constant.DefaultConfigOrder}, @@ -164,8 +164,14 @@ func newTestManager(s store.Store, mockForge *forge_mocks.MockForge) (services.M return services.NewManager(cmd, s, setupForge) } -// startGRPCServer binds to a random TCP port, registers Woodpecker's gRPC -// services, and starts serving. Shutdown happens via t.Cleanup. +// startGRPCServer binds to a random TCP port and serves Woodpecker's gRPC +// services via the shared server_rpc.Serve helper. A fresh prometheus.Registry +// is passed so subtests don't collide on metric names. +// +// Shutdown is synchronous: t.Cleanup cancels the serve context (triggering +// GracefulStop inside Serve) and then blocks until Serve has returned. +// Without this wait, the next subtest can start while the previous server's +// goroutines are still live, which races on shared state like server.Config. func startGRPCServer(ctx context.Context, t *testing.T, s store.Store) string { t.Helper() @@ -173,44 +179,24 @@ func startGRPCServer(ctx context.Context, t *testing.T, s store.Store) string { require.NoError(t, err, "listen on random port for gRPC") addr := lis.Addr().String() - jwtManager := server_rpc.NewJWTManager(TestJWTSecret) - authorizer := server_rpc.NewAuthorizer(jwtManager) - - grpcServer := grpc.NewServer( - grpc.StreamInterceptor(authorizer.StreamInterceptor), - grpc.UnaryInterceptor(authorizer.UnaryInterceptor), - grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ - MinTime: shortTimeout, - }), - ) - - proto.RegisterWoodpeckerServer(grpcServer, server_rpc.NewTestWoodpeckerServer( - server.Config.Services.Scheduler, - server.Config.Services.Logs, - s, - prometheus.NewRegistry(), - )) - proto.RegisterWoodpeckerAuthServer(grpcServer, server_rpc.NewWoodpeckerAuthServer( - jwtManager, - TestAgentToken, - s, - )) - + serveCtx, cancel := context.WithCancelCause(ctx) stopped := make(chan struct{}) - grpcCtx, grpcCancel := context.WithCancelCause(ctx) go func() { - <-grpcCtx.Done() - grpcServer.GracefulStop() - close(stopped) - }() - go func() { - if err := grpcServer.Serve(lis); err != nil { - grpcCancel(err) - } + defer close(stopped) + _ = server_rpc.Serve(serveCtx, server_rpc.ServeConfig{ + Listener: lis, + Store: s, + Scheduler: server.Config.Services.Scheduler, + Logger: server.Config.Services.Logs, + JWTSecret: TestJWTSecret, + AgentToken: TestAgentToken, + KeepaliveMinTime: shortTimeout, + Registerer: prometheus.NewRegistry(), + }) }() t.Cleanup(func() { - grpcCancel(nil) + cancel(nil) <-stopped }) return addr diff --git a/e2e/setup/wait.go b/e2e/setup/wait.go index babe6b632..1786d162f 100644 --- a/e2e/setup/wait.go +++ b/e2e/setup/wait.go @@ -36,6 +36,21 @@ const ( defaultInterval = 100 * time.Millisecond ) +// pollUntil polls condition every defaultInterval until it returns true or +// timeout is exceeded. Returns true if condition was met, false on timeout. +// Callers are expected to handle the timeout case themselves, typically with +// a t.Fatalf that reports the last observed state. +func pollUntil(timeout time.Duration, condition func() bool) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if condition() { + return true + } + time.Sleep(defaultInterval) + } + return condition() +} + // isTerminal returns true if the status is a final (non-running) state. func isTerminal(s model.StatusValue) bool { switch s { @@ -58,23 +73,20 @@ func WaitForPipeline(t *testing.T, s store.Store, pipelineID int64) *model.Pipel func WaitForPipelineStatus(t *testing.T, s store.Store, pipelineID int64, wantStatus model.StatusValue, timeout time.Duration) *model.Pipeline { t.Helper() - deadline := time.Now().Add(timeout) - for time.Now().Before(deadline) { - p, err := s.GetPipeline(pipelineID) + var p *model.Pipeline + ok := pollUntil(timeout, func() bool { + var err error + p, err = s.GetPipeline(pipelineID) require.NoError(t, err, "get pipeline %d", pipelineID) - if wantStatus != "" { - if p.Status == wantStatus { - return p - } - } else if isTerminal(p.Status) { - return p + return p.Status == wantStatus } - - time.Sleep(defaultInterval) + return isTerminal(p.Status) + }) + if ok { + return p } - p, _ := s.GetPipeline(pipelineID) t.Fatalf("timeout waiting for pipeline %d: last status=%q (want %q)", pipelineID, p.Status, wantStatus) return nil } @@ -85,37 +97,32 @@ func WaitForPipelineStatus(t *testing.T, s store.Store, pipelineID int64, wantSt func WaitForAgentRegistered(t *testing.T, s store.Store, agents ...*AgentEnv) { t.Helper() - deadline := time.Now().Add(shortTimeout) - for time.Now().Before(deadline) { - allFound := true + allRegistered := func() bool { for _, env := range agents { if env.AgentID == 0 { - allFound = false - break + return false } if _, err := s.AgentFind(env.AgentID); err != nil { - allFound = false - break + return false } } - if allFound { - // Apply any deferred OrgID patches. - for _, env := range agents { - if env.requestOrgID == model.IDNotSet { - continue - } - agent, err := s.AgentFind(env.AgentID) - require.NoError(t, err, "find agent %d to patch OrgID", env.AgentID) - agent.OrgID = env.requestOrgID - require.NoError(t, s.AgentUpdate(agent), - "patch OrgID on agent %d", env.AgentID) - } - return - } - time.Sleep(defaultInterval) + return true + } + if !pollUntil(shortTimeout, allRegistered) { + t.Fatal("timeout: not all agents registered with the server") } - t.Fatal("timeout: not all agents registered with the server") + // Apply any deferred OrgID patches. + for _, env := range agents { + if env.requestOrgID == model.IDNotSet { + continue + } + agent, err := s.AgentFind(env.AgentID) + require.NoError(t, err, "find agent %d to patch OrgID", env.AgentID) + agent.OrgID = env.requestOrgID + require.NoError(t, s.AgentUpdate(agent), + "patch OrgID on agent %d", env.AgentID) + } } // WaitForStep polls the store until a named step in the given pipeline reaches @@ -132,33 +139,29 @@ func WaitForStep(t *testing.T, s store.Store, pipeline *model.Pipeline, stepName func WaitForStepStatus(t *testing.T, s store.Store, pipeline *model.Pipeline, stepName string, wantState model.StatusValue, timeout time.Duration) *model.Step { t.Helper() - deadline := time.Now().Add(timeout) - for time.Now().Before(deadline) { + var found *model.Step + ok := pollUntil(timeout, func() bool { steps, err := s.StepList(pipeline.ID) require.NoError(t, err, "list steps for pipeline %d", pipeline.ID) - for _, step := range steps { if step.Name != stepName { continue } + found = step if wantState != "" { - if step.State == wantState { - return step - } - } else if isTerminal(step.State) { - return step + return step.State == wantState } + return isTerminal(step.State) } - time.Sleep(defaultInterval) + return false + }) + if ok { + return found } - steps, _ := s.StepList(pipeline.ID) var lastState model.StatusValue - for _, step := range steps { - if step.Name == stepName { - lastState = step.State - break - } + if found != nil { + lastState = found.State } if wantState != "" { t.Fatalf("timeout waiting for step %q in pipeline %d to reach state %q: last state=%q", @@ -200,13 +203,10 @@ func AssertWorkflowRanOnAgent(t *testing.T, s store.Store, pipeline *model.Pipel func WaitForWorkersReady(t *testing.T, q queue.Queue, minWorkers int) { t.Helper() - deadline := time.Now().Add(shortTimeout) - for time.Now().Before(deadline) { - info := q.Info(context.Background()) - if info.Stats.Workers >= minWorkers { - return - } - time.Sleep(defaultInterval) + if pollUntil(shortTimeout, func() bool { + return q.Info(context.Background()).Stats.Workers >= minWorkers + }) { + return } info := q.Info(context.Background()) @@ -221,21 +221,20 @@ func WaitForWorkersReady(t *testing.T, q queue.Queue, minWorkers int) { func WaitForStepRunning(t *testing.T, s store.Store, pipelineID int64, stepName string) { t.Helper() - deadline := time.Now().Add(shortTimeout) - for time.Now().Before(deadline) { + running := pollUntil(shortTimeout, func() bool { p, err := s.GetPipeline(pipelineID) require.NoError(t, err, "get pipeline %d", pipelineID) steps, err := s.StepList(p.ID) require.NoError(t, err, "list steps for pipeline %d", pipelineID) - for _, step := range steps { if step.Name == stepName && step.State == model.StatusRunning { - return + return true } } - time.Sleep(defaultInterval) + return false + }) + if !running { + t.Fatalf("timeout waiting for step %q in pipeline %d to reach StatusRunning", stepName, pipelineID) } - - t.Fatalf("timeout waiting for step %q in pipeline %d to reach StatusRunning", stepName, pipelineID) } diff --git a/server/rpc/serve.go b/server/rpc/serve.go new file mode 100644 index 000000000..fb7754f2c --- /dev/null +++ b/server/rpc/serve.go @@ -0,0 +1,88 @@ +// Copyright 2026 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 rpc + +import ( + "context" + "fmt" + "net" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/rs/zerolog/log" + "google.golang.org/grpc" + "google.golang.org/grpc/keepalive" + + "go.woodpecker-ci.org/woodpecker/v3/rpc/proto" + "go.woodpecker-ci.org/woodpecker/v3/server/logging" + "go.woodpecker-ci.org/woodpecker/v3/server/scheduler" + "go.woodpecker-ci.org/woodpecker/v3/server/store" +) + +// ServeConfig bundles everything Serve needs. Callers build this from a +// *cli.Command (production) or with literals (tests). +type ServeConfig struct { + Listener net.Listener + Store store.Store + Scheduler scheduler.Scheduler + Logger logging.Log + JWTSecret string + AgentToken string + KeepaliveMinTime time.Duration + // Registerer is where the server's prometheus metrics are registered. + // Pass prometheus.DefaultRegisterer in production; pass a fresh + // prometheus.NewRegistry() in tests to avoid duplicate-registration + // panics when the server is created multiple times. + Registerer prometheus.Registerer +} + +// Serve registers Woodpecker's gRPC services on cfg.Listener and blocks +// until ctx is canceled or Serve returns an error. GracefulStop is +// triggered on ctx cancellation. The listener is owned by Serve — it is +// closed when grpc.Server.Serve returns. +func Serve(ctx context.Context, cfg ServeConfig) error { + jwtManager := NewJWTManager(cfg.JWTSecret) + authorizer := NewAuthorizer(jwtManager) + + grpcServer := grpc.NewServer( + grpc.StreamInterceptor(authorizer.StreamInterceptor), + grpc.UnaryInterceptor(authorizer.UnaryInterceptor), + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ + MinTime: cfg.KeepaliveMinTime, + }), + ) + + proto.RegisterWoodpeckerServer(grpcServer, NewWoodpeckerServer( + cfg.Scheduler, cfg.Logger, cfg.Store, cfg.Registerer, + )) + proto.RegisterWoodpeckerAuthServer(grpcServer, NewWoodpeckerAuthServer( + jwtManager, cfg.AgentToken, cfg.Store, + )) + + grpcCtx, cancel := context.WithCancelCause(ctx) + defer cancel(nil) + + go func() { + <-grpcCtx.Done() + log.Info().Msg("terminating grpc service gracefully") + grpcServer.GracefulStop() + log.Info().Msg("grpc service stopped") + }() + + if err := grpcServer.Serve(cfg.Listener); err != nil { + return fmt.Errorf("grpc server failed: %w", err) + } + return nil +} diff --git a/server/rpc/server.go b/server/rpc/server.go index 588852b6a..d9063254f 100644 --- a/server/rpc/server.go +++ b/server/rpc/server.go @@ -36,33 +36,16 @@ type WoodpeckerServer struct { peer RPC } -func NewWoodpeckerServer(scheduler scheduler.Scheduler, logger logging.Log, store store.Store) proto.WoodpeckerServer { - pipelineTime := promauto.NewGaugeVec(prometheus.GaugeOpts{ - Namespace: "woodpecker", - Name: "pipeline_time", - Help: "Pipeline time.", - }, []string{"repo", "branch", "status", "pipeline"}) - pipelineCount := promauto.NewCounterVec(prometheus.CounterOpts{ - Namespace: "woodpecker", - Name: "pipeline_count", - Help: "Pipeline count.", - }, []string{"repo", "branch", "status", "pipeline"}) - peer := RPC{ - store: store, - scheduler: scheduler, - logger: logger, - pipelineTime: pipelineTime, - pipelineCount: pipelineCount, +// NewWoodpeckerServer creates a WoodpeckerServer with its metrics registered +// into registerer. Pass prometheus.DefaultRegisterer in production; pass a +// fresh prometheus.NewRegistry() in tests to avoid "duplicate metrics +// collector registration" panics when the server is created multiple times. +// A nil registerer defaults to prometheus.DefaultRegisterer. +func NewWoodpeckerServer(scheduler scheduler.Scheduler, logger logging.Log, store store.Store, registerer prometheus.Registerer) proto.WoodpeckerServer { + if registerer == nil { + registerer = prometheus.DefaultRegisterer } - return &WoodpeckerServer{peer: peer} -} - -// NewTestWoodpeckerServer creates a WoodpeckerServer for e2e tests. -// It is using a caller-supplied prometheus registry. -// Use this in tests to avoid "duplicate metrics collector registration" panics when the server is created multiple times. -// (promauto in NewWoodpeckerServer registers into the global default registry, which panics on duplicate names). -func NewTestWoodpeckerServer(scheduler scheduler.Scheduler, logger logging.Log, store store.Store, registry *prometheus.Registry) proto.WoodpeckerServer { - factory := promauto.With(registry) + factory := promauto.With(registerer) pipelineTime := factory.NewGaugeVec(prometheus.GaugeOpts{ Namespace: "woodpecker", Name: "pipeline_time",