merge from develop/6

Author: chavakula

.appveyor.yml

@@ -7,7 +7,7 @@ branches:
     - /release\/[0-9.]+/
 
 image:
-  - Visual Studio 2015
+  - Visual Studio 2022
   # - Visual Studio 2017
 
 cache:
@@ -16,7 +16,7 @@ cache:
 environment:
   GOPATH: c:\gopath
   GOROOT: c:\Program Files\Go
-  GOVERSION: 1.21.11
+  GOVERSION: 1.26.0
   GO111MODULE: 'on'
   GOPROXY: 'https://proxy.golang.org'
 
@@ -28,7 +28,7 @@ environment:
 
 install:
   - rmdir c:\go /s /q
-  - appveyor DownloadFile https://storage.googleapis.com/golang/go%GOVERSION%.windows-%GOARCH%.msi
+  - appveyor DownloadFile https://go.dev/dl/go%GOVERSION%.windows-%GOARCH%.msi
   - msiexec /i go%GOVERSION%.windows-%GOARCH%.msi /q
   - echo %PATH%
   - echo %GOPATH%

.circleci/config.yml

@@ -6,7 +6,7 @@ orbs:
 parameters:
   go_version:
     type: string
-    default: "1.21.11"
+    default: "1.26.0"
 
 sensu_go_build_env: &sensu_go_build_env
   docker:

.github/workflows/static-check.yml

@@ -12,17 +12,19 @@ jobs:
   staticcheck:
     name: staticcheck (project)
     runs-on: ubuntu-latest
+    env:
+      GO_VERSION: 1.26.0
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v4
         with:
           fetch-depth: 1
-      - uses: WillAbides/setup-go-faster@v1.14.0
+      - uses: actions/setup-go@v5
         with:
-          go-version: "1.21.x"
+          go-version: "1.26.x"
       - uses: dominikh/staticcheck-action@v1.3.1
         with:
           version: "2025.1.1"
           install-go: false
-          cache-key: "1.21.x"
+          cache-key: "1.26.x"
 env:
-  GO_VERSION: 1.21.3
+  GO_VERSION: 1.26.0

CHANGELOG-6.md

@@ -5,6 +5,44 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 and this project adheres to [Semantic
 Versioning](http://semver.org/spec/v2.0.0.html).
+## [Unreleased]
+
+### Added
+- Added backend-side validation for `runtime_assets` during CheckConfig creation and update.
+  - If any referenced asset does not exist in the namespace, the request will be rejected with a validation error.
+  - This change improves usability by catching missing or invalid assets early, preventing runtime execution failures and reducing debugging overhead.
+- Added `ttl_status` field to `Check` and `CheckConfig` to configure the status (warning or critical) for TTL failure events. This field accepts values 1 (warning) or 2 (critical), with critical (2) as the default to maintain backward compatibility.
+- Added asset lifecycle management with `last-accessed` timestamp tracking to enable intelligent cleanup of unused assets based on usage patterns.
+- Added plugin-based hook system for agent extensions with `BeforeAgentRun` and `AfterAgentRun` lifecycle hooks, supporting priority-based execution ordering and configurable error handling.
+- Added `assets-cleanup-interval` and `assets-cleanup-max-age` configuration options for controlling periodic asset cleanup behavior.
+- Added `FindUnusedAssets()` and `DeleteAsset()` functions for programmatic asset cleanup operations.
+- Added `GetAPIRouter()` and `GetDB()` methods to agent for plugin access to HTTP router and database connection.
+- Added `continue_on_error` field to `core/v2.Pipeline`.
+  This flag controls whether pipeline execution should continue when an error occurs in a handler, filter, mutator, or during asset resolution.
+  Defaults to `false`, preserving existing behavior.
+- Added `ttl_status` field to `Check` and `CheckConfig` to configure the status (warning or critical) for TTL failure events. This field accepts values 1 (warning) or 2 (critical), with warning (1) as the default to maintain backward compatibility.
+- Added Prometheus metrics collection for HTTP API requests, including request count (`sensu_go_http_requests_total`), request duration (`sensu_go_http_request_duration_seconds`), and client error count (`sensu_go_http_client_errors_total`) for better observability of backend API performance.
+
+### Changed
+- Improved asset error handling with new `AgentAssetError` interface providing structured error reporting with exit codes and detailed messages.
+- Enhanced asset manager architecture by restructuring initialization to support pre-initialized database connections and better enterprise integration.
+- Improved command execution timeout error handling for more reliable test execution.
+
+### Fixed
+- Fixed asset retrieval error aggregation to continue processing remaining assets when individual assets fail, using `multierr` for comprehensive error reporting.
+- Upgraded Go version to 1.24.3 across all CI/CD pipelines
+- Updated CircleCI, AppVeyor, and GitHub Actions to use Go 1.24.x
+- Updated minimum Go requirement in README from 1.16 to 1.24
+
+### Fixed
+- Resolved Go 1.24 compatibility issues in agent version and event tests
+- Fixed non-constant format string error in pipeline filter legacy code
+- Updated `GetAssets` method to return an error when one or more required assets are missing.
+  When `continue_on_error` is `false`, missing assets or other errors will stop the pipeline execution (existing behavior).
+  When set to `true`, the pipeline will continue executing subsequent workflows until one succeeds.
+- Modified handler execution behavior to return a non-zero status code on failure.
+  Previously, handler errors were only logged without affecting pipeline.
+
 ## [6.13.1] - 2025-05-28
 
 ### Fixed

README.md

@@ -45,7 +45,7 @@ to get started with one of Sensu's [supported packages](https://docs.sensu.io/se
 ### Building from source
 
 The various components of Sensu Go can be manually built from this repository.
-You will first need [Go >= 1.16](https://golang.org/doc/install#install)
+You will first need [Go >= 1.24](https://golang.org/doc/install#install)
 installed. Then, you should clone this repository **outside** of the GOPATH
 since Sensu Go uses [Go Modules](https://github.com/golang/go/wiki/Modules):
 ```

agent/agent.go

@@ -10,21 +10,23 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"math/rand"
 	"net/http"
 	"net/url"
 	"os"
 	"os/signal"
+	"sort"
 	"strings"
 	"sync"
 	"syscall"
 
 	time "github.com/echlebek/timeproxy"
 	"github.com/gogo/protobuf/proto"
 	"github.com/google/uuid"
+	"github.com/gorilla/mux"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"go.uber.org/multierr"
 	"golang.org/x/time/rate"
 
 	corev2 "github.com/sensu/core/v2"
@@ -38,6 +40,7 @@ import (
 	"github.com/sensu/sensu-go/util/retry"
 	utilstrings "github.com/sensu/sensu-go/util/strings"
 	"github.com/sirupsen/logrus"
+	bolt "go.etcd.io/bbolt"
 )
 
 const (
@@ -134,11 +137,45 @@ func GetDefaultAgentName() string {
 	return defaultAgentName
 }
 
+// Plugin is the hook-based plugin interface.
+type Plugin interface {
+	// Name returns a human-friendly plugin name.
+	Name() string
+
+	// Priority controls ordering; higher runs earlier in BeforeRun and later in AfterRun.
+	// Return 0 for default priority.
+	Priority() int
+
+	// BeforeAgentRun is executed before Agent does its core work.
+	// If it returns an error, Agent decides whether to abort or continue.
+	BeforeAgentRun(ctx context.Context, a *Agent) error
+
+	// AfterAgentRun is executed after Agent's core work (even when core returns error).
+	// It's good for cleanup or metrics; errors should be reported but usually shouldn't abort.
+	AfterAgentRun(ctx context.Context, a *Agent) error
+
+	// IsFatal optional static policy: if true, any error from this plugin is fatal by default
+	// Implementors can return false (default).
+	IsFatal() bool
+}
+
+// BasePlugin small helper so implementers don't need to provide all methods.
+type BasePlugin struct{}
+
+func (BasePlugin) Name() string                                       { return "base" }
+func (BasePlugin) Priority() int                                      { return 0 }
+func (BasePlugin) BeforeAgentRun(ctx context.Context, a *Agent) error { return nil }
+func (BasePlugin) AfterAgentRun(ctx context.Context, a *Agent) error  { return nil }
+func (BasePlugin) IsFatal() bool                                      { return false }
+
 // An Agent receives and acts on messages from a Sensu Backend.
 type Agent struct {
 	allowList          []allowList
 	api                *http.Server
+	apiRouter          *mux.Router
+	dbConn             *bolt.DB
 	assetGetter        asset.Getter
+	plugins            []Plugin
 	backendSelector    BackendSelector
 	config             *Config
 	connected          bool
@@ -207,6 +244,17 @@ func NewAgentContext(ctx context.Context, config *Config) (*Agent, error) {
 		maxSessionLength: config.MaxSessionLength,
 	}
 
+	// Prepare the HTTP API server
+	apiServer, apiRouter := newServer(agent)
+	agent.api = apiServer
+	agent.apiRouter = apiRouter
+	// Prepare the BoltDB before Run
+	db, dbErr := asset.GetBoltDBConnection(agent.config.CacheDir)
+	if dbErr != nil {
+		return nil, dbErr
+	}
+	agent.dbConn = db
+
 	agent.statsdServer = NewStatsdServer(agent)
 	agent.handler.AddHandler(transport.MessageTypeEntityConfig, agent.handleEntityConfig)
 
@@ -224,7 +272,7 @@ func NewAgentContext(ctx context.Context, config *Config) (*Agent, error) {
 		return nil, fmt.Errorf("error creating agent: %s", err)
 	}
 
-	allowList, err := readAllowList(config.AllowList, ioutil.ReadFile)
+	allowList, err := readAllowList(config.AllowList, os.ReadFile)
 	if err != nil {
 		return nil, err
 	}
@@ -239,6 +287,10 @@ func NewAgentContext(ctx context.Context, config *Config) (*Agent, error) {
 	return agent, nil
 }
 
+func (a *Agent) GetConfig() *Config {
+	return a.config
+}
+
 func (a *Agent) sendMessage(msg *transport.Message) {
 	logger.WithFields(logrus.Fields{
 		"type":         msg.Type,
@@ -313,6 +365,11 @@ func (a *Agent) buildTransportHeaderMap() http.Header {
 	return header
 }
 
+// RegisterPlugin registers an agent's plugin
+func (a *Agent) RegisterPlugin(p Plugin) {
+	a.plugins = append(a.plugins, p)
+}
+
 // Run starts the Agent.
 //
 // 1. Start the asset manager.
@@ -332,17 +389,37 @@ func (a *Agent) Run(ctx context.Context) error {
 		}
 	}()
 	defer cancel()
+
+	if len(a.plugins) > 0 {
+		// Keep list sorted descending by Priority (higher first)
+		sort.SliceStable(a.plugins, func(i, j int) bool {
+			return a.plugins[i].Priority() > a.plugins[j].Priority()
+		})
+	}
+
+	var beforeErrs error
+	// Call BeforeAgentRun in order
+	for _, p := range a.plugins {
+		if err := p.BeforeAgentRun(ctx, a); err != nil {
+			bErr := fmt.Errorf("plugin %s BeforeRun: %w", p.Name(), err)
+			beforeErrs = multierr.Append(beforeErrs, bErr)
+			if p.IsFatal() {
+				return beforeErrs
+			}
+		}
+	}
+	if beforeErrs != nil {
+		logger.WithError(beforeErrs).Warn("some plugin BeforeAgentRun errors occurred")
+	}
+
 	a.header = a.buildTransportHeaderMap()
 
 	// Fail the agent after startup if the id is invalid
 	logger.Debug("validating agent name")
 	if err := corev2.ValidateName(a.config.AgentName); err != nil {
 		return fmt.Errorf("invalid agent name: %v", err)
 	}
-	logger.Debug("validating keepalive warning timeout")
-	if timeout := a.config.KeepaliveWarningTimeout; timeout < 5 {
-		return fmt.Errorf("bad keepalive timeout: %d (minimum value is 5 seconds)", timeout)
-	}
+
 	logger.Debug("validating keepalive critical timeout")
 	if timeout := a.config.KeepaliveCriticalTimeout; timeout > 0 && timeout < 5 {
 		return fmt.Errorf("bad keepalive critical timeout: %d (minimum value is 5 seconds)", timeout)
@@ -387,7 +464,7 @@ func (a *Agent) Run(ctx context.Context) error {
 			limit = rate.Limit(asset.DefaultAssetsRateLimit)
 		}
 		var err error
-		a.assetGetter, err = assetManager.StartAssetManager(ctx, rate.NewLimiter(limit, a.config.AssetsBurstLimit))
+		a.assetGetter, err = assetManager.StartAssetManager(ctx, a.dbConn, rate.NewLimiter(limit, a.config.AssetsBurstLimit))
 		if err != nil {
 			return err
 		}
@@ -417,6 +494,20 @@ func (a *Agent) Run(ctx context.Context) error {
 	// Wait for context to complete
 	<-ctx.Done()
 	logger.Info("agent shutting down")
+	// Call AfterAgentRun in reverse order (best effort)
+	var afterErrs error
+	for i := len(a.plugins) - 1; i >= 0; i-- {
+		if err := a.plugins[i].AfterAgentRun(ctx, a); err != nil {
+			aErr := fmt.Errorf("plugin %s AfterRun: %w", a.plugins[i].Name(), err)
+			afterErrs = multierr.Append(afterErrs, aErr)
+			if a.plugins[i].IsFatal() {
+				return afterErrs
+			}
+		}
+	}
+	if afterErrs != nil {
+		logger.WithError(afterErrs).Warn("some plugin AfterAgentRun errors occurred")
+	}
 
 	// Wait for all goroutines to gracefully shutdown, but not too long
 	done := make(chan struct{})
@@ -694,9 +785,6 @@ func (a *Agent) Connected() bool {
 // StartAPI starts the Agent HTTP API. After attempting to start the API, if the
 // HTTP server encounters a fatal error, it will shutdown the rest of the agent.
 func (a *Agent) StartAPI(ctx context.Context) {
-	// Prepare the HTTP API server
-	a.api = newServer(a)
-
 	// Allow Stop() to block until the HTTP server shuts down.
 	a.wg.Add(2)
 
@@ -815,6 +903,24 @@ func (a *Agent) connectWithBackoff(ctx context.Context) (transport.Transport, er
 	return conn, err
 }
 
+// GetDB returns the underlying BoltDB connection used by the agent.
+//
+// The returned pointer must not be closed or replaced by callers.
+// It is intended for read and write operations performed by plugins
+// or internal agent components.
+func (a *Agent) GetDB() *bolt.DB {
+	return a.dbConn
+}
+
+// GetAPIRouter returns the HTTP router used by the agent for registering
+// API endpoints.
+//
+// Plugins or internal agent components may use this router to mount their
+// own routes, handlers, or middleware. Callers should not replace the router instance.
+func (a *Agent) GetAPIRouter() *mux.Router {
+	return a.apiRouter
+}
+
 // GracefulShutdown listens for the SIGINT & SIGTERM signals and cancel the
 // contexts once a signal is received.
 func GracefulShutdown(cancel context.CancelFunc) {

agent/api.go

@@ -30,7 +30,7 @@ type sensuVersion struct {
 }
 
 // newServer returns a new HTTP server
-func newServer(a *Agent) *http.Server {
+func newServer(a *Agent) (*http.Server, *mux.Router) {
 	router := mux.NewRouter()
 	registerRoutes(a, router)
 
@@ -41,7 +41,7 @@ func newServer(a *Agent) *http.Server {
 		ReadTimeout:  15 * time.Second,
 	}
 
-	return server
+	return server, router
 }
 
 func registerRoutes(a *Agent, r *mux.Router) {

agent/api_test.go

@@ -113,7 +113,7 @@ func TestHealthz(t *testing.T) {
 
 func TestVersion(t *testing.T) {
 	var (
-		versionResponse = `{"version":""}`
+		versionResponse = `{"version":"(devel)"}`
 	)
 
 	testCases := []struct {