Add registry inspector to the HTTP server (#50207)

So, we can view and delete state entries via a Web interface.

Author: rdner

auditbeat/auditbeat.reference.yml

@@ -1808,6 +1808,30 @@ logging.files:
 # mutex profile.
 #http.pprof.mutex_profile_rate: 0
 
+# WARNING: Internal debugging tool. Not a supported product feature.
+#
+# This setting is intended for use by Elastic engineers during development
+# and troubleshooting. It has no authentication and may be changed or
+# removed in any release without notice.
+#
+# When enabled, the following endpoints are exposed on the HTTP listener:
+#   GET    /debug/state-inspector/states        JSON dump of every state entry
+#   GET    /debug/state-inspector/states.html   Browser UI for the same
+#   DELETE /debug/state-inspector/states/<key>  Permanently delete a state entry
+#
+# The state store may contain sensitive information such as source file
+# paths, S3 object keys, AWS account identifiers, and source hostnames,
+# all of which will be returned to any client that can reach the endpoint.
+# Deleting state entries can cause duplicate processing, gaps in ingestion,
+# or data loss; for active inputs, deletes may be transient and silently
+# overwritten by the in-memory registrar.
+#
+# If you must enable this for troubleshooting, bind http.host to a loopback
+# address (127.0.0.1), a Unix socket, or a Windows named pipe, and disable
+# it again as soon as you are done.
+#
+#http.debug.state_inspector.enabled: false
+
 # ============================== Process Security ==============================
 
 # Enable or disable seccomp system call filtering on Linux. Default is enabled.

docs/reference/filebeat/http-endpoint.md

@@ -46,11 +46,14 @@ The HTTP endpoint has the following configuration settings:
 `http.pprof.mutex_profile_rate`
 :   (Optional) `mutex_profile_rate` controls the fraction of mutex contention events that are reported in the mutex profile available from `/debug/pprof/mutex`. On average 1/rate events are reported. To turn off profiling entirely, pass rate 0. The default value is 0.
 
+`http.debug.state_inspector.enabled`
+:   (Optional) Enable the state store inspector. **This is an internal debugging tool for Elastic engineers, not a supported product feature.** It has no authentication, may expose sensitive data (file paths, S3 object keys, AWS account identifiers, hostnames), and may be changed or removed in any release without notice. Deleting state entries can cause duplicate processing, gaps in ingestion, or data loss. If you must enable it, bind `http.host` to a loopback address, Unix socket, or Windows named pipe, and disable it again when done. Default is `false`. See [State Inspector](#state-inspector) for details.
+
 This is the list of paths you can access. For pretty JSON output append `?pretty` to the URL.
 
 You can query a unix socket using the `cURL` command and the `--unix-socket` flag.
 
-```js
+```sh
 curl -XGET --unix-socket '/var/run/filebeat.sock' 'http:/stats/?pretty'
 ```
 
@@ -59,11 +62,11 @@ curl -XGET --unix-socket '/var/run/filebeat.sock' 'http:/stats/?pretty'
 
 `/` provides basic info from the Filebeat. Example:
 
-```js
+```sh
 curl -XGET 'localhost:5066/?pretty'
 ```
 
-```js subs=true
+```json subs=true
 {
   "beat": "filebeat",
   "hostname": "example.lan",
@@ -78,11 +81,11 @@ curl -XGET 'localhost:5066/?pretty'
 
 `/stats` reports internal metrics. Example:
 
-```js
+```sh
 curl -XGET 'localhost:5066/stats?pretty'
 ```
 
-```js
+```json
 {
   "beat": {
     "cpu": {
@@ -195,9 +198,37 @@ The actual output may contain more metrics specific to Filebeat
 
 A request may optionally specify a `type` query parameter to request metrics for a specific type of input. And `pretty` may be included to have the returned JSON be pretty formatted.
 
-```js
+```sh
 curl 'http://localhost:5066/inputs/'
 curl 'http://localhost:5066/inputs/?pretty'
 curl 'http://localhost:5066/inputs/?type=aws-s3&pretty'
 ```
 
+
+## State Inspector [state-inspector]
+
+```{applies_to}
+stack: preview 9.5
+```
+
+::::{warning}
+**Internal debugging tool — not a supported product feature.** This is intended for use by Elastic engineers during development and troubleshooting. It has no authentication and may be changed or removed in any release without notice.
+
+The state store may contain sensitive information such as source file paths, S3 object keys, AWS account identifiers, and source hostnames, all of which will be returned to any client that can reach the endpoint. Deleting state entries can cause duplicate processing, gaps in ingestion, or data loss; for active inputs, deletes may be transient and silently overwritten by the in-memory registrar.
+
+If you must enable this for troubleshooting, bind `http.host` to a loopback address (`127.0.0.1`), a Unix socket, or a Windows named pipe, and disable it again as soon as you are done.
+::::
+
+These endpoints are only available when `http.debug.state_inspector.enabled` is set to `true`. Filebeat exposes its input registry through this endpoint.
+
+`/debug/state-inspector/states` returns a JSON array of all key-value pairs in the internal state store. `/debug/state-inspector/states.html` serves a browser UI for the same data.
+
+```sh
+curl -XGET 'localhost:5066/debug/state-inspector/states?pretty'
+```
+
+To permanently delete a specific state entry:
+
+```sh
+curl -XDELETE 'localhost:5066/debug/state-inspector/states/<key>'
+```

filebeat/beater/filebeat.go

@@ -325,6 +325,10 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	}
 	defer stateStore.Close()
 
+	if b.API != nil {
+		b.API.SetStateInspectorRegistry(stateStore.shared.registry, stateStore.storeName)
+	}
+
 	// If notifier is set, configure the listener for output configuration
 	// The notifier passes the elasticsearch output configuration down to the Elasticsearch backed state storage
 	// in order to allow it fully configure

filebeat/filebeat.reference.yml

@@ -3078,6 +3078,30 @@ logging.files:
 # mutex profile.
 #http.pprof.mutex_profile_rate: 0
 
+# WARNING: Internal debugging tool. Not a supported product feature.
+#
+# This setting is intended for use by Elastic engineers during development
+# and troubleshooting. It has no authentication and may be changed or
+# removed in any release without notice.
+#
+# When enabled, the following endpoints are exposed on the HTTP listener:
+#   GET    /debug/state-inspector/states        JSON dump of every state entry
+#   GET    /debug/state-inspector/states.html   Browser UI for the same
+#   DELETE /debug/state-inspector/states/<key>  Permanently delete a state entry
+#
+# The state store may contain sensitive information such as source file
+# paths, S3 object keys, AWS account identifiers, and source hostnames,
+# all of which will be returned to any client that can reach the endpoint.
+# Deleting state entries can cause duplicate processing, gaps in ingestion,
+# or data loss; for active inputs, deletes may be transient and silently
+# overwritten by the in-memory registrar.
+#
+# If you must enable this for troubleshooting, bind http.host to a loopback
+# address (127.0.0.1), a Unix socket, or a Windows named pipe, and disable
+# it again as soon as you are done.
+#
+#http.debug.state_inspector.enabled: false
+
 # ============================== Process Security ==============================
 
 # Enable or disable seccomp system call filtering on Linux. Default is enabled.

heartbeat/heartbeat.reference.yml

@@ -1891,6 +1891,30 @@ logging.files:
 # mutex profile.
 #http.pprof.mutex_profile_rate: 0
 
+# WARNING: Internal debugging tool. Not a supported product feature.
+#
+# This setting is intended for use by Elastic engineers during development
+# and troubleshooting. It has no authentication and may be changed or
+# removed in any release without notice.
+#
+# When enabled, the following endpoints are exposed on the HTTP listener:
+#   GET    /debug/state-inspector/states        JSON dump of every state entry
+#   GET    /debug/state-inspector/states.html   Browser UI for the same
+#   DELETE /debug/state-inspector/states/<key>  Permanently delete a state entry
+#
+# The state store may contain sensitive information such as source file
+# paths, S3 object keys, AWS account identifiers, and source hostnames,
+# all of which will be returned to any client that can reach the endpoint.
+# Deleting state entries can cause duplicate processing, gaps in ingestion,
+# or data loss; for active inputs, deletes may be transient and silently
+# overwritten by the in-memory registrar.
+#
+# If you must enable this for troubleshooting, bind http.host to a loopback
+# address (127.0.0.1), a Unix socket, or a Windows named pipe, and disable
+# it again as soon as you are done.
+#
+#http.debug.state_inspector.enabled: false
+
 # ============================== Process Security ==============================
 
 # Enable or disable seccomp system call filtering on Linux. Default is enabled.

libbeat/_meta/config/http.reference.yml.tmpl

@@ -38,3 +38,27 @@
 # Controls the fraction of mutex contention events that are reported in the
 # mutex profile.
 #http.pprof.mutex_profile_rate: 0
+
+# WARNING: Internal debugging tool. Not a supported product feature.
+#
+# This setting is intended for use by Elastic engineers during development
+# and troubleshooting. It has no authentication and may be changed or
+# removed in any release without notice.
+#
+# When enabled, the following endpoints are exposed on the HTTP listener:
+#   GET    /debug/state-inspector/states        JSON dump of every state entry
+#   GET    /debug/state-inspector/states.html   Browser UI for the same
+#   DELETE /debug/state-inspector/states/<key>  Permanently delete a state entry
+#
+# The state store may contain sensitive information such as source file
+# paths, S3 object keys, AWS account identifiers, and source hostnames,
+# all of which will be returned to any client that can reach the endpoint.
+# Deleting state entries can cause duplicate processing, gaps in ingestion,
+# or data loss; for active inputs, deletes may be transient and silently
+# overwritten by the in-memory registrar.
+#
+# If you must enable this for troubleshooting, bind http.host to a loopback
+# address (127.0.0.1), a Unix socket, or a Windows named pipe, and disable
+# it again as soon as you are done.
+#
+#http.debug.state_inspector.enabled: false

libbeat/api/config.go

@@ -19,13 +19,24 @@ package api
 
 import "os"
 
+// StateInspectorConfig holds the configuration for the state store inspector.
+type StateInspectorConfig struct {
+	Enabled bool `config:"enabled"`
+}
+
+// DebugConfig holds the configuration for debugging endpoints.
+type DebugConfig struct {
+	StateInspector StateInspectorConfig `config:"state_inspector"`
+}
+
 // Config is the configuration for the API endpoint.
 type Config struct {
-	Enabled            bool   `config:"enabled"`
-	Host               string `config:"host"`
-	Port               int    `config:"port"`
-	User               string `config:"named_pipe.user"`
-	SecurityDescriptor string `config:"named_pipe.security_descriptor"`
+	Enabled            bool        `config:"enabled"`
+	Host               string      `config:"host"`
+	Port               int         `config:"port"`
+	User               string      `config:"named_pipe.user"`
+	SecurityDescriptor string      `config:"named_pipe.security_descriptor"`
+	Debug              DebugConfig `config:"debug"`
 }
 
 // DefaultConfig is the default configuration used by the API endpoint.

libbeat/api/server.go

@@ -26,6 +26,8 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/elastic/beats/v7/libbeat/statestore"
+	"github.com/elastic/beats/v7/libbeat/statestore/inspector"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 )
@@ -49,6 +51,7 @@ type Server struct {
 	mutex      sync.Mutex
 	httpServer *http.Server
 	state      serverState
+	inspector  *inspector.Handler
 }
 
 // New creates a new API Server with no routes attached.
@@ -150,6 +153,26 @@ func (s *Server) AttachHandler(route string, h http.Handler) (err error) {
 	return nil
 }
 
+// AttachStateInspector creates and registers the state store inspector
+// handler if enabled in config. Calling it more than once or when the
+// inspector is disabled is a no-op.
+func (s *Server) AttachStateInspector() error {
+	if !s.config.Debug.StateInspector.Enabled || s.inspector != nil {
+		return nil
+	}
+	s.inspector = inspector.New()
+	return s.AttachHandler("/debug/state-inspector/", http.StripPrefix("/debug/state-inspector", s.inspector))
+}
+
+// SetStateInspectorRegistry provides the backing registry and store name to
+// the state inspector. Each HTTP request will obtain its own Store instance
+// via registry.Get(name). This is a no-op when the inspector is not enabled.
+func (s *Server) SetStateInspectorRegistry(registry *statestore.Registry, name string) {
+	if s.inspector != nil {
+		s.inspector.SetRegistry(registry, name)
+	}
+}
+
 // Router returns the mux.Router that handles all request to the server.
 func (s *Server) Router() *http.ServeMux {
 	return s.mux

libbeat/cmd/instance/beat.go

@@ -466,6 +466,9 @@ func (b *Beat) launch(settings Settings, bt beat.Creator) error {
 				return fmt.Errorf("failed to attach http handlers for pprof: %w", err)
 			}
 		}
+		if err := b.API.AttachStateInspector(); err != nil {
+			return fmt.Errorf("failed to attach state inspector: %w", err)
+		}
 	}
 
 	// Do not load seccomp for osquerybeat, it was disabled before V2 in the configuration file