func (d *Daemon) NotifyChange(ctx context.Context, change v9.Change) error {
	switch change.Kind {
	case v9.GitChange:
		gitUpdate := change.Source.(v9.GitUpdate)
		if d.Repo.Origin().Equivalent(gitUpdate.URL) && gitUpdate.Branch != d.GitConfig.Branch {
			d.Logger.Log("msg", "notified about unrelated change",
				"url", gitUpdate.URL,
				"branch", gitUpdate.Branch)
			break
		}
		d.Repo.Notify()
	case v9.ImageChange:
		imageUpdate := change.Source.(v9.ImageUpdate)
		d.ImageRefresh <- imageUpdate.Name
	}
	return nil
}
Tell the daemon to synchronise the cluster with the manifests in
the git repo. This has an error return value because upstream there
may be comms difficulties or other sources of problems; here, we
always succeed because it's just bookkeeping.
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
}
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
	}
}
func (d *Daemon) queueJob(do jobFunc) job.ID {
	id := job.ID(guid.New())
	enqueuedAt := time.Now()
	d.Jobs.Enqueue(&job.Job{
		ID: id,
		Do: func(logger log.Logger) error {
			queueDuration.Observe(time.Since(enqueuedAt).Seconds())
			_, err := d.executeJob(id, do, logger)
			if err != nil {
				return err
			}
			return nil
		},
	})
	queueLength.Set(float64(d.Jobs.Len()))
	d.JobStatusCache.SetStatus(id, job.Status{StatusString: job.StatusQueued})
	return id
}
func (d *Daemon) UpdateManifests(ctx context.Context, spec update.Spec) (job.ID, error) {
	var id job.ID
	if spec.Type == "" {
		return id, errors.New("no type in update spec")
	}
	switch s := spec.Spec.(type) {
	case release.Changes:
		if s.ReleaseKind() == update.ReleaseKindPlan {
			id := job.ID(guid.New())
			_, err := d.executeJob(id, d.makeJobFromUpdate(d.release(spec, s)), d.Logger)
			return id, err
		}
		return d.queueJob(d.makeLoggingJobFunc(d.makeJobFromUpdate(d.release(spec, s)))), nil
	case resource.PolicyUpdates:
		return d.queueJob(d.makeLoggingJobFunc(d.makeJobFromUpdate(d.updatePolicies(spec, s)))), nil
	case update.ManualSync:
		return d.queueJob(d.sync()), nil
	default:
		return id, fmt.Errorf(`unknown update type "%s"`, spec.Type)
	}
}
func (d *Daemon) sync() jobFunc {
	return func(ctx context.Context, jobID job.ID, logger log.Logger) (job.Result, error) {
		var result job.Result
		ctx, cancel := context.WithTimeout(ctx, d.SyncTimeout)
		defer cancel()
		err := d.Repo.Refresh(ctx)
		if err != nil {
			return result, err
		}
		head, err := d.Repo.BranchHead(ctx)
		if err != nil {
			return result, err
		}
		if d.GitVerifySignaturesMode != sync.VerifySignaturesModeNone {
			var latestValidRev string
			if latestValidRev, _, err = latestValidRevision(ctx, d.Repo, d.SyncState, d.GitVerifySignaturesMode); err != nil {
				return result, err
			} else if head != latestValidRev {
				result.Revision = latestValidRev
				return result, fmt.Errorf(
					"The branch HEAD in the git repo is not verified, and fluxd is unable to sync to it. The last verified commit was %.8s. HEAD is %.8s.",
					latestValidRev,
					head,
				)
			}
		}
		result.Revision = head
		return result, err
	}
}
func (d *Daemon) updatePolicies(spec update.Spec, updates resource.PolicyUpdates) updateFunc {
	return func(ctx context.Context, jobID job.ID, working *git.Checkout, logger log.Logger) (job.Result, error) {
		var workloadIDs []resource.ID
		result := job.Result{
			Spec:   &spec,
			Result: update.Result{},
		}
		var anythingAutomated bool
		for workloadID, u := range updates {
			if d.Cluster.IsAllowedResource(workloadID) {
				result.Result[workloadID] = update.WorkloadResult{
					Status: update.ReleaseStatusSkipped,
				}
			}
			if u.Add.Has(policy.Automated) {
				anythingAutomated = true
			}
			cm, err := d.getManifestStore(working)
			if err != nil {
				return result, err
			}
			updated, err := cm.UpdateWorkloadPolicies(ctx, workloadID, u)
			if err != nil {
				result.Result[workloadID] = update.WorkloadResult{
					Status: update.ReleaseStatusFailed,
					Error:  err.Error(),
				}
				switch err := err.(type) {
				case manifests.StoreError:
					result.Result[workloadID] = update.WorkloadResult{
						Status: update.ReleaseStatusFailed,
						Error:  err.Error(),
					}
				default:
					return result, err
				}
			}
			if !updated {
				result.Result[workloadID] = update.WorkloadResult{
					Status: update.ReleaseStatusSkipped,
				}
			} else {
				workloadIDs = append(workloadIDs, workloadID)
				result.Result[workloadID] = update.WorkloadResult{
					Status: update.ReleaseStatusSuccess,
				}
			}
		}
		if len(workloadIDs) == 0 {
			return result, nil
		}
		commitAuthor := ""
		if d.GitConfig.SetAuthor {
			commitAuthor = spec.Cause.User
		}
		commitAction := git.CommitAction{
			Author:  commitAuthor,
			Message: policyCommitMessage(updates, spec.Cause),
		}
		if err := working.CommitAndPush(ctx, commitAction, &note{JobID: jobID, Spec: spec}, d.ManifestGenerationEnabled); err != nil {
			d.AskForSync()
			return result, err
		}
		if anythingAutomated {
			d.AskForAutomatedWorkloadImageUpdates()
		}
		var err error
		result.Revision, err = working.HeadRevision(ctx)
		if err != nil {
			return result, err
		}
		return result, nil
	}
}
func (d *Daemon) release(spec update.Spec, c release.Changes) updateFunc {
	return func(ctx context.Context, jobID job.ID, working *git.Checkout, logger log.Logger) (job.Result, error) {
		var zero job.Result
		rs, err := d.getManifestStore(working)
		if err != nil {
			return zero, err
		}
		rc := release.NewReleaseContext(d.Cluster, rs, d.Registry)
		result, err := release.Release(ctx, rc, c, logger)
		if err != nil {
			return zero, err
		}
		var revision string
		if c.ReleaseKind() == update.ReleaseKindExecute {
			commitMsg := spec.Cause.Message
			if commitMsg == "" {
				commitMsg = c.CommitMessage(result)
			}
			commitAuthor := ""
			if d.GitConfig.SetAuthor {
				commitAuthor = spec.Cause.User
			}
			commitAction := git.CommitAction{
				Author:  commitAuthor,
				Message: commitMsg,
			}
			if err := working.CommitAndPush(ctx, commitAction, &note{JobID: jobID, Spec: spec, Result: result}, d.ManifestGenerationEnabled); err != nil {
				d.Repo.Notify()
				return zero, err
			}
			revision, err = working.HeadRevision(ctx)
			if err != nil {
				return zero, err
			}
		}
		return job.Result{
			Revision: revision,
			Spec:     &spec,
			Result:   result,
		}, nil
	}
}