Quick tidy up

Author: nick-jones

internal/auditlog/tail.go

@@ -16,6 +16,8 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+// Tail streams audit log entries relating to fluxcd resources. Only audit log entries relating to non-system users
+// patching or creating resources are returned.
 func Tail(ctx context.Context, projectID string, clusterName string, cb func(*audit.AuditLog) error) error {
 	client, err := logging.NewClient(ctx)
 	if err != nil {

internal/config/config.go

@@ -2,11 +2,13 @@ package config
 
 import (
 	"fmt"
+	"io"
 	"os"
 
 	"gopkg.in/yaml.v3"
 )
 
+// Config is the application configuration
 type Config struct {
 	GoogleCloudProjectID string `yaml:"googleCloudProjectId"`
 	GKEClusterName       string `yaml:"gkeClusterName"`
@@ -20,15 +22,21 @@ type Config struct {
 	} `yaml:"notification"`
 }
 
-func Parse(path string) (Config, error) {
+// ParseFile parses configuration from a given file path
+func ParseFile(path string) (Config, error) {
 	f, err := os.Open(path)
 	if err != nil {
 		return Config{}, fmt.Errorf("failed to open config file: %w", err)
 	}
 	defer f.Close()
 
+	return Parse(f)
+}
+
+// Parse parses configuration by reading the contents of the supplied reader
+func Parse(r io.Reader) (Config, error) {
 	var config Config
-	if err = yaml.NewDecoder(f).Decode(&config); err != nil {
+	if err := yaml.NewDecoder(r).Decode(&config); err != nil {
 		return Config{}, fmt.Errorf("failed to parse config file: %w", err)
 	}
 	return config, nil

internal/datastore/badgerdb.go

@@ -11,19 +11,24 @@ import (
 	"github.com/e-flux-platform/fluxcd-suspend-notifier/internal/k8s"
 )
 
+// ErrNotFound is returned when an entry cannot be found in the underlying store
 var ErrNotFound = errors.New("not found")
 
+// Store is a basic badgerdb backed persistence mechanism
 type Store struct {
 	db *badger.DB
 }
 
+// Entry represents a single item held by the store. It relates to a single resource reference, and holds information
+// about its suspension status.
 type Entry struct {
 	Resource  k8s.ResourceReference `json:"resource"`
 	Suspended bool                  `json:"suspended"`
 	UpdatedBy string                `json:"updatedBy"`
 	UpdatedAt time.Time             `json:"updatedAt"`
 }
 
+// NewBadgerStore instantiates a Store instance. Data will be persisted the directory pointed at by the supplied path.
 func NewBadgerStore(path string) (*Store, error) {
 	if path == "" {
 		return nil, errors.New("badger store path cannot be empty")
@@ -37,6 +42,7 @@ func NewBadgerStore(path string) (*Store, error) {
 	}, nil
 }
 
+// GetEntry retrieves an entry
 func (s *Store) GetEntry(resource k8s.ResourceReference) (Entry, error) {
 	var entry Entry
 	err := s.db.View(func(txn *badger.Txn) error {
@@ -59,6 +65,7 @@ func (s *Store) GetEntry(resource k8s.ResourceReference) (Entry, error) {
 	return entry, err
 }
 
+// SaveEntry creates or replaces an entry
 func (s *Store) SaveEntry(entry Entry) error {
 	return s.db.Update(func(txn *badger.Txn) error {
 		data, err := json.Marshal(entry)
@@ -69,6 +76,7 @@ func (s *Store) SaveEntry(entry Entry) error {
 	})
 }
 
+// Close cleans up any underlying resources
 func (s *Store) Close() error {
 	return s.db.Close()
 }

internal/fluxcd/resource.go

@@ -1,5 +1,7 @@
 package fluxcd
 
+// Resource represents an abstract suspendable fluxcd resource. Only the fields relevant to this application are
+// covered here
 type Resource struct {
 	Metadata struct {
 		Name      string `json:"name"`
@@ -10,6 +12,7 @@ type Resource struct {
 	} `json:"spec"`
 }
 
+// ResourceList represents a list of resources, aligned to how this would be presented by the kubernetes API
 type ResourceList struct {
 	Items []Resource `json:"items"`
 }

internal/k8s/client.go

@@ -13,11 +13,14 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 )
 
+// Client is a thing wrapper around the kubernetes client. It exposes functions relevant for checking what fluxcd
+// resources exist, and what their underlying state look like.
 type Client struct {
 	client       *kubernetes.Clientset
 	apiExtClient *clientset.Clientset
 }
 
+// NewClient instantiates and returns a Client
 func NewClient(configPath string) (*Client, error) {
 	var (
 		config *rest.Config
@@ -48,6 +51,7 @@ func NewClient(configPath string) (*Client, error) {
 	}, nil
 }
 
+// GetRawResource retrieves a raw resource from the kubernetes API. This is used to fetch fluxcd custom resources.
 func (c *Client) GetRawResource(ctx context.Context, resource ResourceReference) ([]byte, error) {
 	absPath := path.Join(
 		"apis",
@@ -66,6 +70,7 @@ func (c *Client) GetRawResource(ctx context.Context, resource ResourceReference)
 	return body, nil
 }
 
+// GetRawResources retrieves a list of raw resources
 func (c *Client) GetRawResources(ctx context.Context, group ResourceType) ([]byte, error) {
 	absPath := path.Join(
 		"apis",
@@ -81,6 +86,7 @@ func (c *Client) GetRawResources(ctx context.Context, group ResourceType) ([]byt
 	return body, nil
 }
 
+// GetCustomResourceDefinitions fetches all custom resource definitions registered with the cluster.
 func (c *Client) GetCustomResourceDefinitions(ctx context.Context, listOptions metav1.ListOptions) (*v1.CustomResourceDefinitionList, error) {
 	return c.apiExtClient.
 		ApiextensionsV1().

internal/k8s/resource.go

@@ -5,18 +5,21 @@ import (
 	"strings"
 )
 
+// ResourceType represents a kubernetes resource type
 type ResourceType struct {
 	Group   string `json:"group"`
 	Version string `json:"version"`
 	Kind    string `json:"kind"`
 }
 
+// ResourceReference represents a reference to a kubernetes resource instance
 type ResourceReference struct {
 	Type      ResourceType `json:"type"`
 	Namespace string       `json:"namespace"`
 	Name      string       `json:"name"`
 }
 
+// ResourceReferenceFromPath parses an API path to a resource reference
 func ResourceReferenceFromPath(path string) (ResourceReference, error) {
 	parts := strings.Split(path, "/")
 	if len(parts) != 6 {

internal/notification/filtering.go

@@ -8,6 +8,7 @@ import (
 	"github.com/expr-lang/expr/vm"
 )
 
+// NewFilteringNotifier instantiates and returns FilteringNotifier
 func NewFilteringNotifier(rawExpr string, delegate Notifier) (*FilteringNotifier, error) {
 	filter, err := expr.Compile(rawExpr)
 	if err != nil {
@@ -19,11 +20,13 @@ func NewFilteringNotifier(rawExpr string, delegate Notifier) (*FilteringNotifier
 	}, nil
 }
 
+// FilteringNotifier is a notifier implementation that filters notifications via an expression.
 type FilteringNotifier struct {
 	filter   *vm.Program
 	delegate Notifier
 }
 
+// Notify passes the notification to the underlying delegate if the expression is satisfied.
 func (fn *FilteringNotifier) Notify(ctx context.Context, notif Notification) error {
 	env := map[string]interface{}{
 		"resource":  notif.Resource,

internal/notification/multi.go

@@ -2,16 +2,19 @@ package notification
 
 import "context"
 
+// MultiNotifier is used to broadcast notifications
 type MultiNotifier struct {
 	notifiers []Notifier
 }
 
+// NewMultiNotifier instantiates and returns MultiNotifier
 func NewMultiNotifier(notifiers []Notifier) *MultiNotifier {
 	return &MultiNotifier{
 		notifiers: notifiers,
 	}
 }
 
+// Notify passes the notification to all underlying notifiers
 func (an *MultiNotifier) Notify(ctx context.Context, notif Notification) error {
 	for _, n := range an.notifiers {
 		if err := n.Notify(ctx, notif); err != nil {

internal/notification/notifier.go

@@ -6,13 +6,15 @@ import (
 	"github.com/e-flux-platform/fluxcd-suspend-notifier/internal/k8s"
 )
 
+// Notification carries information relevant for dispatching external notifications
 type Notification struct {
 	Resource             k8s.ResourceReference
 	Suspended            bool
 	Email                string
 	GoogleCloudProjectID string
 }
 
+// Notifier is the interface that is expected to be implemented for notification mechanisms
 type Notifier interface {
 	Notify(context.Context, Notification) error
 }

internal/notification/slack.go

@@ -11,11 +11,13 @@ import (
 	"time"
 )
 
+// SlackNotifier sends notifications to Slack via a webhook
 type SlackNotifier struct {
 	client     *http.Client
 	webhookURL string
 }
 
+// NewSlackNotifier instantiates and returns SlackNotifier
 func NewSlackNotifier(webhookURL string) (*SlackNotifier, error) {
 	if webhookURL == "" {
 		return nil, errors.New("empty webhook url supplied")
@@ -28,10 +30,12 @@ func NewSlackNotifier(webhookURL string) (*SlackNotifier, error) {
 	}, nil
 }
 
+// SlackWebhook is a Slack webhook payload
 type SlackWebhook struct {
 	Attachments []SlackAttachment `json:"attachments,omitempty"`
 }
 
+// SlackAttachment forms part of a Slack webhook payload
 type SlackAttachment struct {
 	Color      string                 `json:"color"`
 	AuthorName string                 `json:"author_name"`
@@ -40,12 +44,14 @@ type SlackAttachment struct {
 	Fields     []SlackAttachmentField `json:"fields,omitempty"`
 }
 
+// SlackAttachmentField forms part of a Slack webhook attachment value
 type SlackAttachmentField struct {
 	Title string `json:"title"`
 	Value string `json:"value"`
 	Short bool   `json:"short"`
 }
 
+// Notify sends a notification via the underlying Slack webhook URL.
 func (sn *SlackNotifier) Notify(ctx context.Context, notif Notification) error {
 	var (
 		action string

internal/watch/watcher.go

@@ -20,6 +20,8 @@ import (
 	"github.com/e-flux-platform/fluxcd-suspend-notifier/internal/notification"
 )
 
+// Watcher is used to orchestrate notifications. It discovers fluxcd resources, watches for changes, and notifies when
+// the suspension status changes.
 type Watcher struct {
 	googleCloudProjectID string
 	gkeClusterName       string
@@ -28,6 +30,7 @@ type Watcher struct {
 	notifier             notifier
 }
 
+// NewWatcher instantiates and returns Watcher
 func NewWatcher(
 	googleCloudProjectID string,
 	gkeClusterName string,
@@ -59,6 +62,8 @@ type notifier interface {
 	Notify(context.Context, notification.Notification) error
 }
 
+// Watch blocks waiting for fluxcd resource suspension statuses to change. When a change is observed, the notifier
+// is invoked.
 func (w *Watcher) Watch(ctx context.Context) error {
 	resourceTypes, err := w.resolveFluxResourceTypes(ctx)
 	if err != nil {
@@ -72,6 +77,7 @@ func (w *Watcher) Watch(ctx context.Context) error {
 	return w.watch(ctx, resourceTypes)
 }
 
+// resolveFluxResourceTypes returns fluxcd resource types; specifically only those that can be suspended.
 func (w *Watcher) resolveFluxResourceTypes(ctx context.Context) ([]k8s.ResourceType, error) {
 	crds, err := w.k8sClient.GetCustomResourceDefinitions(ctx, metav1.ListOptions{
 		LabelSelector: "app.kubernetes.io/part-of=flux",
@@ -97,6 +103,9 @@ func (w *Watcher) resolveFluxResourceTypes(ctx context.Context) ([]k8s.ResourceT
 	return types, nil
 }
 
+// init retries the suspension status of all fluxcd resource instances that are a suspendable resource type. This is
+// useful when starting from scratch, to build an initial picture. Equally, if the application has been down for a
+// period of time, it allows for the state to be synchronised.
 func (w *Watcher) init(ctx context.Context, types []k8s.ResourceType) error {
 	slog.Info("initializing")
 	seen := make(map[string]struct{})
@@ -131,6 +140,8 @@ func (w *Watcher) init(ctx context.Context, types []k8s.ResourceType) error {
 	return nil
 }
 
+// watch tails audit logs, waiting for modifications to fluxcd resource types that are suspendable. When a modification
+// is observed, the resource state is evaluated via processResource
 func (w *Watcher) watch(ctx context.Context, types []k8s.ResourceType) error {
 	slog.Info("watching for resource modifications")
 
@@ -171,6 +182,8 @@ func (w *Watcher) watch(ctx context.Context, types []k8s.ResourceType) error {
 	})
 }
 
+// processResource checks to see if the suspend status has been modified. If it has, a notification is dispatched. If
+// the resource has never been seen before, we simply save the state.
 func (w *Watcher) processResource(
 	ctx context.Context,
 	resourceRef k8s.ResourceReference,

main.go

@@ -31,7 +31,7 @@ func run(ctx context.Context) error {
 		return errors.New("config path environment variable not set")
 	}
 
-	conf, err := config.Parse(configPath)
+	conf, err := config.ParseFile(configPath)
 	if err != nil {
 		return fmt.Errorf("failed to parse config: %w", err)
 	}