Add APIs for computing status based on fetching resource info from a

cluster

kstatus/wait/wait.go

@@ -0,0 +1,258 @@
+// Copyright 2019 The Kubernetes Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package wait
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/pkg/errors"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/kustomize/kstatus/status"
+)
+
+// ResourceIdentifier defines the functions needed to identify
+// a resource in a cluster. This interface is implemented by
+// both unstructured.Unstructured and the standard Kubernetes types.
+type ResourceIdentifier interface {
+	GetName() string
+	GetNamespace() string
+	GetAPIVersion() string
+	GetKind() string
+}
+
+// Resolver provides the functions for resolving status of a list of resources.
+type Resolver struct {
+	// DynamicClient is the client used to talk
+	// with the cluster
+	client client.Reader
+
+	// statusComputeFunc defines which function should be used for computing
+	// the status of a resource. This is available for testing purposes.
+	statusComputeFunc func(u *unstructured.Unstructured) (*status.Result, error)
+
+	// pollInterval defines the frequency with which the resolver should poll
+	// the cluster for the state of resources. More frequent polling will
+	// lead to more load on the cluster.
+	pollInterval time.Duration
+}
+
+// NewResolver creates a new resolver with the provided client. Fetching
+// and polling of resources will be done using the provided client.
+func NewResolver(client client.Reader, pollInterval time.Duration) *Resolver {
+	return &Resolver{
+		client:            client,
+		statusComputeFunc: status.Compute,
+		pollInterval:      pollInterval,
+	}
+}
+
+// ResourceResult is the status result for a given resource. It provides
+// information about the resource if the request was successful and an
+// error if something went wrong.
+type ResourceResult struct {
+	Result *status.Result
+
+	Resource ResourceIdentifier
+
+	Error error
+}
+
+// FetchAndResolve returns the status for a list of resources. It will return
+// the status for each of them individually. The slice of ResourceIdentifiers will
+// only be used to get the information needed to fetch the updated state of
+// the resources from the cluster.
+func (r *Resolver) FetchAndResolve(ctx context.Context, resources []ResourceIdentifier) []ResourceResult {
+	var results []ResourceResult
+
+	for _, resource := range resources {
+		u, err := r.fetchResource(ctx, resource)
+		if err != nil {
+			if k8serrors.IsNotFound(errors.Cause(err)) {
+				results = append(results, ResourceResult{
+					Resource: resource,
+					Result: &status.Result{
+						Status:  status.CurrentStatus,
+						Message: "Resource does not exist",
+					},
+				})
+			} else {
+				results = append(results, ResourceResult{
+					Result: &status.Result{
+						Status:  status.UnknownStatus,
+						Message: fmt.Sprintf("Error fetching resource from cluster: %v", err),
+					},
+					Resource: resource,
+					Error:    err,
+				})
+			}
+			continue
+		}
+		res, err := r.statusComputeFunc(u)
+		results = append(results, ResourceResult{
+			Result:   res,
+			Resource: resource,
+			Error:    err,
+		})
+	}
+
+	return results
+}
+
+// Event is returned through the channel returned after a call
+// to WaitForStatus. It contains an update to either an individual
+// resource or to the aggregate status for the set of resources.
+type Event struct {
+	// Type defines which type of event this is.
+	Type EventType
+
+	// AggregateStatus is the aggregated status for all the provided resources.
+	AggregateStatus status.Status
+
+	// EventResource is information about the event to which this event pertains.
+	// This is only populated for ResourceUpdate events.
+	EventResource *EventResource
+}
+
+type EventType string
+
+const (
+	// The status/message for a resource has changed. This also means the
+	// aggregate status might have changed.
+	ResourceUpdate EventType = "ResourceUpdate"
+
+	// All resources have reached the current status.
+	Completed EventType = "Completed"
+
+	// The wait was stopped before all resources could reach the
+	// Current status.
+	Aborted EventType = "Aborted"
+)
+
+// EventResource contains information about the resource for which
+// a specific Event pertains.
+type EventResource struct {
+	// Identifier contains information that identifies which resource
+	// this information is about.
+	Identifier ResourceIdentifier
+
+	// Status is the latest status for the given resource.
+	Status status.Status
+
+	// Message is more details about the status.
+	Message string
+
+	// Error is set if there was a problem identifying the status
+	// of the resource. For example, if polling the cluster for information
+	// about the resource failed.
+	Error error
+}
+
+// WaitForStatus polls all the provided resources until all of them has
+// reached the Current status. Updates the channel as resources change their status and
+// when the wait is either completed or aborted.
+func (r *Resolver) WaitForStatus(ctx context.Context, resources []ResourceIdentifier) <-chan Event {
+	eventChan := make(chan Event)
+
+	go func() {
+		ticker := time.NewTicker(r.pollInterval)
+
+		defer func() {
+			ticker.Stop()
+			// Make sure the channel is closed so consumers can detect that
+			// we have completed.
+			close(eventChan)
+		}()
+
+		// No need to wait if we have no resources. We consider
+		// this a situation where the status is Current.
+		if len(resources) == 0 {
+			eventChan <- Event{
+				Type:            Completed,
+				AggregateStatus: status.CurrentStatus,
+				EventResource:   nil,
+			}
+			return
+		}
+
+		// Initiate a new waitStatus object to keep track of the
+		// resources while polling the state.
+		waitState := newWaitState(resources, r.statusComputeFunc)
+
+		// Loop until either all resources have reached the Current status
+		// or until the wait is cancelled through the context. In both cases
+		// we will break out of the loop by returning from the function.
+		for {
+			select {
+			case <-ctx.Done():
+				// The context has been cancelled, so report the most recent
+				// aggregate status, report it through the channel and then
+				// break out of the loop (which will close the channel).
+				eventChan <- Event{
+					Type:            Aborted,
+					AggregateStatus: waitState.AggregateStatus(),
+				}
+				return
+			case <-ticker.C:
+				// Every time the ticker fires, fetch all resources from the cluster,
+				// check if their status has changed and send an event for each resource
+				// with a new status. In each event, we also include the latest aggregate
+				// status. Finally, if the aggregate status becomes Current, send a final
+				// Completed type event and then return.
+				for id := range waitState.ResourceWaitStates {
+					// Make sure we have a local copy since we are passing
+					// pointers to this variable as parameters to functions
+					identifier := id
+					u, err := r.fetchResource(ctx, &identifier)
+					eventResource, updateObserved := waitState.ResourceObserved(&identifier, u, err)
+					// Find the aggregate status based on the new state for this resource.
+					aggStatus := waitState.AggregateStatus()
+					// We want events for changes in status for each resource, so send
+					// an event for this resource before checking if the aggregate status
+					// has become Current.
+					if updateObserved {
+						eventChan <- Event{
+							Type:            ResourceUpdate,
+							AggregateStatus: aggStatus,
+							EventResource:   &eventResource,
+						}
+					}
+					// If aggregate status is Current, we are done!
+					if aggStatus == status.CurrentStatus {
+						eventChan <- Event{
+							Type:            Completed,
+							AggregateStatus: status.CurrentStatus,
+						}
+						return
+					}
+				}
+			}
+		}
+	}()
+
+	return eventChan
+}
+
+// fetchResource gets the resource given by the identifier from the cluster
+// through the client available in the Resolver. It returns the resource
+// as an Unstructured.
+func (r *Resolver) fetchResource(ctx context.Context, identifier ResourceIdentifier) (*unstructured.Unstructured, error) {
+	key := types.NamespacedName{Name: identifier.GetName(), Namespace: identifier.GetNamespace()}
+	u := &unstructured.Unstructured{
+		Object: map[string]interface{}{
+			"apiVersion": identifier.GetAPIVersion(),
+			"kind":       identifier.GetKind(),
+		},
+	}
+	err := r.client.Get(ctx, key, u)
+	//return u, err
+	if err != nil {
+		return nil, errors.Wrap(err, "error fetching resource from cluster")
+	}
+	return u, nil
+}