func (d *Daemon) makeLoggingJobFunc(f jobFunc) jobFunc {
	return func(ctx context.Context, id job.ID, logger log.Logger) (job.Result, error) {
		started := time.Now().UTC()
		result, err := f(ctx, id, logger)
		if err != nil {
			return result, err
		}
		logger.Log("revision", result.Revision)
		if result.Revision != "" {
			var workloadIDs []resource.ID
			for id, result := range result.Result {
				if result.Status == update.ReleaseStatusSuccess {
					workloadIDs = append(workloadIDs, id)
				}
			}
			metadata := &event.CommitEventMetadata{
				Revision: result.Revision,
				Spec:     result.Spec,
				Result:   result.Result,
			}
			return result, d.LogEvent(event.Event{
				ServiceIDs: workloadIDs,
				Type:       event.EventCommit,
				StartedAt:  started,
				EndedAt:    started,
				LogLevel:   event.LogLevelInfo,
				Metadata:   metadata,
			})
		}
		return result, nil
	}
}
makeLoggingFunc takes a jobFunc and returns a jobFunc that will log
a commit event with the result.
package daemon
import (
	"bytes"
	"context"
	"fmt"
	"sort"
	"strings"
	"time"
	"github.com/go-kit/kit/log"
	"github.com/pkg/errors"
	"github.com/fluxcd/flux/pkg/api"
	"github.com/fluxcd/flux/pkg/api/v10"
	"github.com/fluxcd/flux/pkg/api/v11"
	"github.com/fluxcd/flux/pkg/api/v6"
	"github.com/fluxcd/flux/pkg/api/v9"
	"github.com/fluxcd/flux/pkg/cluster"
	"github.com/fluxcd/flux/pkg/event"
	"github.com/fluxcd/flux/pkg/git"
	"github.com/fluxcd/flux/pkg/guid"
	"github.com/fluxcd/flux/pkg/image"
	"github.com/fluxcd/flux/pkg/job"
	"github.com/fluxcd/flux/pkg/manifests"
	"github.com/fluxcd/flux/pkg/policy"
	"github.com/fluxcd/flux/pkg/registry"
	"github.com/fluxcd/flux/pkg/release"
	"github.com/fluxcd/flux/pkg/resource"
	"github.com/fluxcd/flux/pkg/sync"
	"github.com/fluxcd/flux/pkg/update"
)
type Daemon struct {
	V                         string
	Cluster                   cluster.Cluster
	Manifests                 manifests.Manifests
	Registry                  registry.Registry
	ImageRefresh              chan image.Name
	Repo                      *git.Repo
	GitConfig                 git.Config
	Jobs                      *job.Queue
	JobStatusCache            *job.StatusCache
	EventWriter               event.EventWriter
	Logger                    log.Logger
	ManifestGenerationEnabled bool
	GitSecretEnabled          bool
	*LoopVars
}
var _ api.Server = &Daemon{}
func (d *Daemon) Version(ctx context.Context) (string, error) {
	return d.V, nil
}
func (d *Daemon) Ping(ctx context.Context) error {
	return d.Cluster.Ping()
}
func (d *Daemon) Export(ctx context.Context) ([]byte, error) {
	return d.Cluster.Export(ctx)
}
type repo interface {
	Dir() string
}
func (d *Daemon) getManifestStore(r repo) (manifests.Store, error) {
	absPaths := git.MakeAbsolutePaths(r, d.GitConfig.Paths)
	if d.ManifestGenerationEnabled {
		return manifests.NewConfigAware(r.Dir(), absPaths, d.Manifests, d.SyncTimeout)
	}
	return manifests.NewRawFiles(r.Dir(), absPaths, d.Manifests), nil
}
func (d *Daemon) getResources(ctx context.Context) (map[string]resource.Resource, v6.ReadOnlyReason, error) {
	var resources map[string]resource.Resource
	var globalReadOnly v6.ReadOnlyReason
	err := d.WithReadonlyClone(ctx, func(checkout *git.Export) error {
		cm, err := d.getManifestStore(checkout)
		if err != nil {
			return err
		}
		resources, err = cm.GetAllResourcesByID(ctx)
		return err
	})
	_, notReady := err.(git.NotReadyError)
	switch {
	case notReady:
		globalReadOnly = v6.ReadOnlyNotReady
	case err == git.ErrNoConfig:
		globalReadOnly = v6.ReadOnlyNoRepo
	case err != nil:
		return nil, globalReadOnly, manifestLoadError(err)
	default:
		globalReadOnly = v6.ReadOnlyMissing
	}
	return resources, globalReadOnly, nil
}
func (d *Daemon) ListServices(ctx context.Context, namespace string) ([]v6.ControllerStatus, error) {
	return d.ListServicesWithOptions(ctx, v11.ListServicesOptions{Namespace: namespace})
}
func (d *Daemon) ListServicesWithOptions(ctx context.Context, opts v11.ListServicesOptions) ([]v6.ControllerStatus, error) {
	if opts.Namespace != "" && len(opts.Services) > 0 {
		return nil, errors.New("cannot filter by 'namespace' and 'workloads' at the same time")
	}
	var clusterWorkloads []cluster.Workload
	var err error
	if len(opts.Services) > 0 {
		clusterWorkloads, err = d.Cluster.SomeWorkloads(ctx, opts.Services)
	} else {
		clusterWorkloads, err = d.Cluster.AllWorkloads(ctx, opts.Namespace)
	}
	if err != nil {
		return nil, errors.Wrap(err, "getting workloads from cluster")
	}
	resources, missingReason, err := d.getResources(ctx)
	if err != nil {
		return nil, err
	}
	var res []v6.ControllerStatus
	for _, workload := range clusterWorkloads {
		readOnly := v6.ReadOnlyOK
		repoIsReadonly := d.Repo.Readonly()
		var policies policy.Set
		if resource, ok := resources[workload.ID.String()]; ok {
			policies = resource.Policies()
		}
		switch {
		case policies == nil:
			readOnly = missingReason
		case repoIsReadonly:
			readOnly = v6.ReadOnlyROMode
		case workload.IsSystem:
			readOnly = v6.ReadOnlySystem
		}
		var syncError string
		if workload.SyncError != nil {
			syncError = workload.SyncError.Error()
		}
		res = append(res, v6.ControllerStatus{
			ID:         workload.ID,
			Containers: containers2containers(workload.ContainersOrNil()),
			ReadOnly:   readOnly,
			Status:     workload.Status,
			Rollout:    workload.Rollout,
			SyncError:  syncError,
			Antecedent: workload.Antecedent,
			Labels:     workload.Labels,
			Automated:  policies.Has(policy.Automated),
			Locked:     policies.Has(policy.Locked),
			Ignore:     policies.Has(policy.Ignore),
			Policies:   policies.ToStringMap(),
		})
	}
	return res, nil
}
type clusterContainers []cluster.Workload
func (cs clusterContainers) Len() int {
	return len(cs)
}
func (cs clusterContainers) Containers(i int) []resource.Container {
	return cs[i].ContainersOrNil()
}
func (d *Daemon) ListImages(ctx context.Context, spec update.ResourceSpec) ([]v6.ImageStatus, error) {
	return d.ListImagesWithOptions(ctx, v10.ListImagesOptions{Spec: spec})
}
func (d *Daemon) ListImagesWithOptions(ctx context.Context, opts v10.ListImagesOptions) ([]v6.ImageStatus, error) {
	if opts.Namespace != "" && opts.Spec != update.ResourceSpecAll {
		return nil, errors.New("cannot filter by 'namespace' and 'workload' at the same time")
	}
	var workloads []cluster.Workload
	var err error
	if opts.Spec != update.ResourceSpecAll {
		id, err := opts.Spec.AsID()
		if err != nil {
			return nil, errors.Wrap(err, "treating workload spec as ID")
		}
		workloads, err = d.Cluster.SomeWorkloads(ctx, []resource.ID{id})
		if err != nil {
			return nil, errors.Wrap(err, "getting some workloads")
		}
	} else {
		workloads, err = d.Cluster.AllWorkloads(ctx, opts.Namespace)
		if err != nil {
			return nil, errors.Wrap(err, "getting all workloads")
		}
	}
	resources, _, err := d.getResources(ctx)
	if err != nil {
		return nil, err
	}
	imageRepos, err := update.FetchImageRepos(d.Registry, clusterContainers(workloads), d.Logger)
	if err != nil {
		return nil, errors.Wrap(err, "getting images for workloads")
	}
	var res []v6.ImageStatus
	for _, workload := range workloads {
		workloadContainers, err := getWorkloadContainers(workload, imageRepos, resources[workload.ID.String()], opts.OverrideContainerFields)
		if err != nil {
			return nil, err
		}
		res = append(res, v6.ImageStatus{
			ID:         workload.ID,
			Containers: workloadContainers,
		})
	}
	return res, nil
}
type jobFunc func(ctx context.Context, jobID job.ID, logger log.Logger) (job.Result, error)
type updateFunc func(ctx context.Context, jobID job.ID, working *git.Checkout, logger log.Logger) (job.Result, error)
func (d *Daemon) makeJobFromUpdate(update updateFunc) jobFunc {
	return func(ctx context.Context, jobID job.ID, logger log.Logger) (job.Result, error) {
		var result job.Result
		err := d.WithWorkingClone(ctx, func(working *git.Checkout) error {
			var err error
			if err = verifyWorkingRepo(ctx, d.Repo, working, d.SyncState, d.GitVerifySignaturesMode); d.GitVerifySignaturesMode != sync.VerifySignaturesModeNone && err != nil {
				return err
			}
			result, err = update(ctx, jobID, working, logger)
			if err != nil {
				return err
			}
			return nil
		})
		if err != nil {
			return result, err
		}
		return result, nil
	}
}
func (d *Daemon) executeJob(id job.ID, do jobFunc, logger log.Logger) (job.Result, error) {
	ctx, cancel := context.WithTimeout(context.Background(), d.SyncTimeout)
	defer cancel()
	d.JobStatusCache.SetStatus(id, job.Status{StatusString: job.StatusRunning})
	result, err := do(ctx, id, logger)
	if err != nil {
		d.JobStatusCache.SetStatus(id, job.Status{StatusString: job.StatusFailed, Err: err.Error(), Result: result})
		return result, err
	}
	d.JobStatusCache.SetStatus(id, job.Status{StatusString: job.StatusSucceeded, Result: result})
	return result, nil
}