aggregation

Documentation

Overview

Package aggregation is responsible for hosting an HTTP server which aggregates results from all of the nodes that are running sonobuoy agent. It is not responsible for dispatching the nodes (see pkg/dispatch), only expecting their results.

Index

• Constants
• func Cleanup(client kubernetes.Interface, plugins []plugin.Interface)
• func GetPatch(annotation string) map[string]interface{}
• func GlobalResultURL(baseURL, pluginName string) (string, error)
• func NewHandler(resultsCallback func(*plugin.Result, http.ResponseWriter)) http.Handler
• func NodeResultURL(baseURL, nodeName, pluginName string) (string, error)
• func Run(client kubernetes.Interface, plugins []plugin.Interface, ...) error
• type Aggregator
  • func NewAggregator(outputDir string, expected []plugin.ExpectedResult) *Aggregator
  • func (a *Aggregator) HandleHTTPResult(result *plugin.Result, w http.ResponseWriter)
  • func (a *Aggregator) IngestResults(ctx context.Context, resultsCh <-chan *plugin.Result)
  • func (a *Aggregator) RunAndMonitorPlugin(ctx context.Context, p plugin.Interface, client kubernetes.Interface, ...)
  • func (a *Aggregator) Wait(stop chan bool)
• type Handler
• type PluginStatus
• type Status
  • func GetStatus(client kubernetes.Interface, namespace string) (*Status, error)

Constants

const (
	// RunningStatus means the sonobuoy run is still in progress.
	RunningStatus string = "running"
	// CompleteStatus means the sonobuoy run is complete.
	CompleteStatus string = "complete"
	// PostProcessingStatus means the plugins are complete. The state is not
	// put in the more finalized, complete, status until any postprocessing is
	// done.
	PostProcessingStatus string = "post-processing"
	// FailedStatus means one or more plugins has failed and the run will not complete successfully.
	FailedStatus string = "failed"
)

const (
	StatusAnnotationName = "sonobuoy.hept.io/status"
	StatusPodName        = "sonobuoy"
)

Functions

func Cleanup

func Cleanup(client kubernetes.Interface, plugins []plugin.Interface)

Cleanup calls cleanup on all plugins

func GetPatch

func GetPatch(annotation string) map[string]interface{}

GetPatch takes a json encoded string and creates a map which can be used as a patch to indicate the Sonobuoy status.

func GlobalResultURL

func GlobalResultURL(baseURL, pluginName string) (string, error)

GlobalResultURL is the URL that results that are not node-specific. Takes the baseURL (http[s]://hostname:port/, with trailing slash) pluginName, and an optional extension. If multiple extensions are provided, only the first one is used.

func NewHandler

func NewHandler(resultsCallback func(*plugin.Result, http.ResponseWriter)) http.Handler

NewHandler constructs a new aggregation handler which will handler results and pass them to the given results callback.

func NodeResultURL

func NodeResultURL(baseURL, nodeName, pluginName string) (string, error)

NodeResultURL is the URL for results for a given node result. Takes the baseURL (http[s]://hostname:port/, with trailing slash) nodeName, pluginName, and an optional extension. If multiple extensions are provided, only the first one is used.

func Run

func Run(client kubernetes.Interface, plugins []plugin.Interface, cfg plugin.AggregationConfig, namespace, outdir string) error

Run runs an aggregation server and gathers results, in accordance with the given sonobuoy configuration.

Basic workflow:

1. Create the aggregator object (`aggr`) to keep track of results
2. Launch the HTTP server with the aggr's HandleHTTPResult function as the callback
3. Run all the aggregation plugins, monitoring each one in a goroutine, configuring them to send failure results through a shared channel
4. Hook the shared monitoring channel up to aggr's IngestResults() function
5. Block until aggr shows all results accounted for (results come in through the HTTP callback), stopping the HTTP server on completion

Types

type Aggregator

type Aggregator struct {
	// OutputDir is the directory to write the node results
	OutputDir string

	// Results stores a map of check-in results the server has seen
	Results map[string]*plugin.Result

	// ExpectedResults stores a map of results the server should expect
	ExpectedResults map[string]*plugin.ExpectedResult

	// FailedResults is a map to track which plugin results were received
	// but returned errors during processing. This enables us to retry results
	// that failed to process if the client tries, as opposed to rejecting
	// them as duplicates. Important if connection resets or network issues
	// are common.
	FailedResults map[string]time.Time
	// contains filtered or unexported fields
}

Aggregator is responsible for taking results from an HTTP server (configured elsewhere), saving them to the filesystem, and keeping track of what has been seen so far, so that we can return when all expected results are present and accounted for.

func NewAggregator

func NewAggregator(outputDir string, expected []plugin.ExpectedResult) *Aggregator

NewAggregator constructs a new Aggregator object to write the given result set out to the given output directory.

func (*Aggregator) HandleHTTPResult

func (a *Aggregator) HandleHTTPResult(result *plugin.Result, w http.ResponseWriter)

HandleHTTPResult is called every time the HTTP server gets a well-formed request with results. This method is responsible for returning with things like a 409 conflict if a node has checked in twice (or a 403 forbidden if a node isn't expected), as well as actually calling handleResult to write the results to OutputDir.

func (*Aggregator) IngestResults

func (a *Aggregator) IngestResults(ctx context.Context, resultsCh <-chan *plugin.Result)

IngestResults takes a channel of results and handles them as they come in. Since most plugins submit over HTTP, this method is currently only used to consume an error stream from each plugin's Monitor() function.

If we support plugins that are just simple commands that the sonobuoy master runs, those plugins can submit results through the same channel.

func (*Aggregator) RunAndMonitorPlugin

func (a *Aggregator) RunAndMonitorPlugin(ctx context.Context, p plugin.Interface, client kubernetes.Interface, nodes []corev1.Node, address string, cert *tls.Certificate)

RunAndMonitorPlugin will start a plugin then monitor it for errors starting/running. Errors detected will be handled by saving an error result in the aggregator.Results.

func (*Aggregator) Wait

func (a *Aggregator) Wait(stop chan bool)

Wait blocks until all expected results have come in.

type Handler

type Handler struct {
	mux.Router
	// ResultsCallback is the function that is called when a result is checked in.
	ResultsCallback func(*plugin.Result, http.ResponseWriter)
}

Handler is a net/http Handler that can handle API requests for aggregation of results from nodes, calling the provided callback with the results

type PluginStatus

type PluginStatus struct {
	Plugin string `json:"plugin"`
	Node   string `json:"node"`
	Status string `json:"status"`
}

PluginStatus represents the current status of an individual plugin.

type Status

type Status struct {
	Plugins []PluginStatus `json:"plugins"`
	Status  string         `json:"status"`
}

Status represents the current status of a Sonobuoy run. TODO(EKF): Find a better name for this struct/package.

func GetStatus

func GetStatus(client kubernetes.Interface, namespace string) (*Status, error)

GetStatus returns the current status status on the sonobuoy pod. If the pod does not exist, is not running, or is missing the status annotation, an error is returned.