// NewAllocReconciler creates a new reconciler that should be used to determine
// the changes required to bring the cluster state inline with the declared jobspec
165987
func NewAllocReconciler(logger log.Logger, allocUpdateFn allocUpdateType, batch bool,
	jobID string, job *structs.Job, deployment *structs.Deployment,
	existingAllocs []*structs.Allocation, taintedNodes map[string]*structs.Node, evalID string) *allocReconciler {
	return &allocReconciler{
		logger:         logger.Named("reconciler"),
		allocUpdateFn:  allocUpdateFn,
		batch:          batch,
		jobID:          jobID,
		job:            job,
		deployment:     deployment.Copy(),
		existingAllocs: existingAllocs,
		taintedNodes:   taintedNodes,
		evalID:         evalID,
		now:            time.Now(),
		result: &reconcileResults{
			desiredTGUpdates:     make(map[string]*structs.DesiredUpdates),
			desiredFollowupEvals: make(map[string][]*structs.Evaluation),
		},
	}
}
791947
func (w *deploymentWatcher) handleAllocUpdate(allocs []*structs.AllocListStub) (allocUpdateResult, error) {
	var res allocUpdateResult

	// Get the latest evaluation index
	latestEval, err := w.jobEvalStatus()
	if err != nil {
		if err == context.Canceled || w.ctx.Err() == context.Canceled {
			return res, err
		}

		return res, fmt.Errorf("failed to determine last evaluation index for job %q: %v", w.j.ID, err)
	}

	deployment := w.getDeployment()
	for _, alloc := range allocs {
		dstate, ok := deployment.TaskGroups[alloc.TaskGroup]
		if !ok {
			continue
		}

		// Determine if the update stanza for this group is progress based
		progressBased := dstate.ProgressDeadline != 0

		// Check if the allocation has failed and we need to mark it for allow
		// replacements
		if progressBased && alloc.DeploymentStatus.IsUnhealthy() &&
			deployment.Active() && !alloc.DesiredTransition.ShouldReschedule() {
			res.allowReplacements = append(res.allowReplacements, alloc.ID)
			continue
		}

		// We need to create an eval so the job can progress.
		if alloc.DeploymentStatus.IsHealthy() && alloc.DeploymentStatus.ModifyIndex > latestEval {
			res.createEval = true
		}

		// If the group is using a progress deadline, we don't have to do anything.
		if progressBased {
			continue
		}

		// Fail on the first bad allocation
		if alloc.DeploymentStatus.IsUnhealthy() {
			// Check if the group has autorevert set
			if dstate.AutoRevert {
				res.rollback = true
			}

			// Since we have an unhealthy allocation, fail the deployment
			res.failDeployment = true
		}

		// All conditions have been hit so we can break
		if res.createEval && res.failDeployment && res.rollback {
			break
		}
	}

	return res, nil
}
588233
func (s *Server) reconcileJobSummaries() error {
	index, err := s.fsm.state.LatestIndex()
	if err != nil {
		return fmt.Errorf("unable to read latest index: %v", err)
	}
	s.logger.Debug("leader reconciling job summaries", "index", index)

	args := &structs.GenericResponse{}
	msg := structs.ReconcileJobSummariesRequestType | structs.IgnoreUnknownTypeFlag
	if _, _, err = s.raftApply(msg, args); err != nil {
		return fmt.Errorf("reconciliation of job summaries failed: %v", err)
	}

	return nil
}
71490
func NewWorker(config Config) (RestoreStatusWorker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	restoreStatus, err := config.RestoreInfoWatcher.RestoreStatus()
	if err != nil {
		return nil, errors.Trace(err)
	}
	w := &restoreWorker{
		config: config,
		status: restoreStatus,
	}
	if err := catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: w.loop,
	}); err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
558631
func newReliableSender(cfg *reliableSenderCfg) *reliableSender {
	return &reliableSender{
		cfg:         *cfg,
		activePeers: make(map[[33]byte]peerManager),
		quit:        make(chan struct{}),
	}
}
936886
func NewCluster(options *Options) (*Cluster, error) {
    cluster := &Cluster{
	nodes: make(map[string]*redisNode),
	connTimeout: options.ConnTimeout,
	readTimeout: options.ReadTimeout,
	writeTimeout: options.WriteTimeout,
	keepAlive: options.KeepAlive,
	aliveTime: options.AliveTime,
	updateList: make(chan updateMesg),
    }

    for i := range options.StartNodes {
	node := &redisNode{
	    address: options.StartNodes[i],
	    connTimeout: options.ConnTimeout,
	    readTimeout: options.ReadTimeout,
	    writeTimeout: options.WriteTimeout,
	    keepAlive: options.KeepAlive,
	    aliveTime: options.AliveTime,
	}

	err := cluster.update(node)
	if err != nil {
	    continue
	} else {
	    go cluster.handleUpdate()
	    return cluster, nil
	}
    }

    return nil, fmt.Errorf("NewCluster: no valid node in %v", options.StartNodes)
}
800572
func NewCertificateUpdater(config Config) worker.Worker {
	return legacy.NewNotifyWorker(&CertificateUpdater{
		addressWatcher:  config.AddressWatcher,
		configGetter:    config.ControllerConfigGetter,
		hostPortsGetter: config.APIHostPortsGetter,
		getter:          config.StateServingInfoGetter,
		setter:          config.StateServingInfoSetter,
	})
}
248205
func NewCoordinator() Coordinator {
	return Coordinator{
		consumerWorker: new(sync.WaitGroup),
		producerWorker: new(sync.WaitGroup),
		state:          coordinatorStateConfigure,
	}
}
1027231
func NewMockClusterRoleInterface(ctrl *gomock.Controller) *MockClusterRoleInterface {
	mock := &MockClusterRoleInterface{ctrl: ctrl}
	mock.recorder = &MockClusterRoleInterfaceMockRecorder{mock}
	return mock
}
608921
func NewAllocRunner(config *Config) (*allocRunner, error) {
	alloc := config.Alloc
	tg := alloc.Job.LookupTaskGroup(alloc.TaskGroup)
	if tg == nil {
		return nil, fmt.Errorf("failed to lookup task group %q", alloc.TaskGroup)
	}

	ar := &allocRunner{
		id:                       alloc.ID,
		alloc:                    alloc,
		clientConfig:             config.ClientConfig,
		consulClient:             config.Consul,
		vaultClient:              config.Vault,
		tasks:                    make(map[string]*taskrunner.TaskRunner, len(tg.Tasks)),
		waitCh:                   make(chan struct{}),
		destroyCh:                make(chan struct{}),
		shutdownCh:               make(chan struct{}),
		state:                    &state.State{},
		stateDB:                  config.StateDB,
		stateUpdater:             config.StateUpdater,
		taskStateUpdatedCh:       make(chan struct{}, 1),
		taskStateUpdateHandlerCh: make(chan struct{}),
		allocUpdatedCh:           make(chan *structs.Allocation, 1),
		deviceStatsReporter:      config.DeviceStatsReporter,
		prevAllocWatcher:         config.PrevAllocWatcher,
		prevAllocMigrator:        config.PrevAllocMigrator,
		devicemanager:            config.DeviceManager,
		driverManager:            config.DriverManager,
	}

	// Create the logger based on the allocation ID
	ar.logger = config.Logger.Named("alloc_runner").With("alloc_id", alloc.ID)

	// Create alloc broadcaster
	ar.allocBroadcaster = cstructs.NewAllocBroadcaster(ar.logger)

	// Create alloc dir
	ar.allocDir = allocdir.NewAllocDir(ar.logger, filepath.Join(config.ClientConfig.AllocDir, alloc.ID))

	// Initialize the runners hooks.
	ar.initRunnerHooks()

	// Create the TaskRunners
	if err := ar.initTaskRunners(tg.Tasks); err != nil {
		return nil, err
	}

	return ar, nil
}
1210328
func NewReprovider(ctx context.Context, rsys routing.ContentRouting, keyProvider KeyChanFunc) *Reprovider {
	return &Reprovider{
		ctx:     ctx,
		trigger: make(chan doneFunc),

		rsys:        rsys,
		keyProvider: keyProvider,
	}
}
490286
func (b *AllocUpdateBatcher) batcher() {
	var timerCh <-chan time.Time
	allocs := make(map[string]*structs.DesiredTransition)
	evals := make(map[string]*structs.Evaluation)
	future := NewBatchFuture()
	for {
		select {
		case <-b.ctx.Done():
			return
		case w := <-b.workCh:
			if timerCh == nil {
				timerCh = time.After(b.batch)
			}

			// Store the eval and alloc updates, and attach the future
			evals[w.e.DeploymentID] = w.e
			for id, upd := range w.allocs {
				allocs[id] = upd
			}

			w.f <- future
		case <-timerCh:
			// Capture the future and create a new one
			f := future
			future = NewBatchFuture()

			// Shouldn't be possible
			if f == nil {
				panic("no future")
			}

			// Create the request
			req := &structs.AllocUpdateDesiredTransitionRequest{
				Allocs: allocs,
				Evals:  make([]*structs.Evaluation, 0, len(evals)),
			}

			for _, e := range evals {
				req.Evals = append(req.Evals, e)
			}

			// Upsert the evals in a go routine
			go f.Set(b.raft.UpdateAllocDesiredTransition(req))

			// Reset the evals list and timer
			evals = make(map[string]*structs.Evaluation)
			allocs = make(map[string]*structs.DesiredTransition)
			timerCh = nil
		}
	}
}
40325
func NewCronJobInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer {
	return NewFilteredCronJobInformer(client, namespace, resyncPeriod, indexers, nil)
}
697211
func NewPeriodicReconciler(interval time.Duration, recFunc func(), eStream EventStream) PeriodicReconciler {
	return &reconciler{
		ival:    interval,
		rFunc:   recFunc,
		eStream: eStream,
		clock:   clockwork.NewRealClock(),
	}
}
325185
func NewFilteredCronJobInformer(client kubernetes.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer {
	return cache.NewSharedIndexInformer(
		&cache.ListWatch{
			ListFunc: func(options v1.ListOptions) (runtime.Object, error) {
				if tweakListOptions != nil {
					tweakListOptions(&options)
				}
				return client.BatchV2alpha1().CronJobs(namespace).List(options)
			},
			WatchFunc: func(options v1.ListOptions) (watch.Interface, error) {
				if tweakListOptions != nil {
					tweakListOptions(&options)
				}
				return client.BatchV2alpha1().CronJobs(namespace).Watch(options)
			},
		},
		&batchv2alpha1.CronJob{},
		resyncPeriod,
		indexers,
	)
}
952768
func NewAllocUpdateBatcher(batchDuration time.Duration, raft DeploymentRaftEndpoints, ctx context.Context) *AllocUpdateBatcher {
	b := &AllocUpdateBatcher{
		batch:  batchDuration,
		raft:   raft,
		ctx:    ctx,
		workCh: make(chan *updateWrapper, 10),
	}

	go b.batcher()
	return b
}
1266224
func NewAlter(ctx *plan.Context, p *plan.Alter) *Alter {
	m := &Alter{
		TaskBase: NewTaskBase(ctx),
		p:        p,
	}
	return m
}