// Run will retrieve scraper targets from the target storage,
// and publish them to nats job queue for gather.
352114
func (s *Scheduler) Run(ctx context.Context) error {
	go func(s *Scheduler, ctx context.Context) {
		for {
			select {
			case <-ctx.Done():
				return
			case <-time.After(s.Interval): // TODO: change to ticker because of garbage collection
				s.gather <- struct{}{}
			}
		}
	}(s, ctx)
	return s.run(ctx)
}
630403
func (c *controller) Run(stopCh <-chan struct{}) {
	defer utilruntime.HandleCrash()
	go func() {
		<-stopCh
		c.config.Queue.Close()
	}()
	r := NewReflector(
		c.config.ListerWatcher,
		c.config.ObjectType,
		c.config.Queue,
		c.config.FullResyncPeriod,
	)
	r.ShouldResync = c.config.ShouldResync
	r.clock = c.clock

	c.reflectorMutex.Lock()
	c.reflector = r
	c.reflectorMutex.Unlock()

	var wg wait.Group
	defer wg.Wait()

	wg.StartWithChannel(stopCh, r.Run)

	wait.Until(c.processLoop, time.Second, stopCh)
}
739233
func (n *Ingester) Run() error {
	pwg := sync.WaitGroup{}
	for i := 0; i < n.ParseConcurrency; i++ {
		pwg.Add(1)
		go func() {
			defer pwg.Done()
			var recordErr error
			for {
				// Source
				var rec interface{}
				rec, recordErr = n.src.Record()
				if recordErr != nil {
					break
				}
				n.Stats.Count("ingest.Record", 1, 1)

				// Parse
				val, err := n.parser.Parse(rec)
				if err != nil {
					n.Log.Printf("couldn't parse record %s, err: %v", rec, err)
					n.Stats.Count("ingest.ParseError", 1, 1)
					continue
				}
				n.Stats.Count("ingest.Parse", 1, 1)

				// Transform
				for _, tr := range n.Transformers {
					err := tr.Transform(val)
					if err != nil {
						n.Log.Printf("Problem with transformer %#v: %v", tr, err)
						n.Stats.Count("ingest.TransformError", 1, 1)
					}
				}
				n.Stats.Count("ingest.Transform", 1, 1)

				// Map
				pr, err := n.mapper.Map(val)
				if err != nil {
					n.Log.Printf("couldn't map val: %s, err: %v", val, err)
					n.Stats.Count("ingest.MapError", 1, 1)
					continue
				}

				// Index
				n.Stats.Count("ingest.Map", 1, 1)
				for _, row := range pr.Rows {
					if n.AllowedFields == nil || n.AllowedFields[row.Field] {
						n.indexer.AddColumn(row.Field, pr.Col, row.ID)
						n.Stats.Count("ingest.AddBit", 1, 1)
					}
				}
				for _, val := range pr.Vals {
					if n.AllowedFields == nil || n.AllowedFields[val.Field] {
						n.indexer.AddValue(val.Field, pr.Col, val.Value)
						n.Stats.Count("ingest.AddValue", 1, 1)
					}
				}
			}
			if recordErr != io.EOF && recordErr != nil {
				n.Log.Printf("error in ingest run loop: %v", recordErr)
			}
		}()
	}
	pwg.Wait()
	return n.indexer.Close()
}
1050930
func (o *Organizer) Run() {
	if o.only != nil {
		o.runTest(o.only)
		return
	}

	for _, test := range o.tests {
		o.runTest(&test)
	}
}
497876
func (e *PGPKeyfinder) Run(m libkb.MetaContext) error {
	e.loadUsers(m)
	e.loadKeys(m)
	return e.runerr
}
1068675
func (c *Copier) Run() {
	for src, w := range c.srcs {
		c.copyJobs.Add(1)
		go c.copySrc(src, w)
	}
}
676779
func Run(sOpts *Options, nOpts *natsd.Options) (*StanServer, error) {
	return RunServerWithOpts(sOpts, nOpts)
}
980116
func (e *MirrorSecrets) Run(c *fi.Context) error {
	return fi.DefaultDeltaRunMethod(e, c)
}
231424
func run(ctx context.Context) (err error) {
	// Open the bucket.
	bucket, err := getBucket(ctx)
	if err != nil {
		err = fmt.Errorf("getBucket: %v", err)
		return
	}

	// Stat the object.
	o, err := getObject(ctx, bucket)
	if err != nil {
		err = fmt.Errorf("getObject: %v", err)
		return
	}

	// Make reads.
	results, err := runWorkers(ctx, o, bucket)
	if err != nil {
		err = fmt.Errorf("runWorkers: %v", err)
		return
	}

	// Print information about the results.
	describeResults(results)

	return
}
1210487
func (o *ImageLookupOptions) Run() error {
	b := o.Builder().
		WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
		LocalParam(o.Local).
		ContinueOnError().
		NamespaceParam(o.Namespace).DefaultNamespace().
		FilenameParam(o.ExplicitNamespace, &o.FilenameOptions).
		Flatten()

	switch {
	case o.Local:
		// perform no lookups on the server
		// TODO: discovery still requires a running server, doesn't fall back correctly
	case len(o.Args) == 0 && len(o.Filenames) == 0:
		b = b.
			LabelSelectorParam(o.Selector).
			SelectAllParam(o.All).
			ResourceTypes("imagestreams")
	case o.List:
		b = b.
			LabelSelectorParam(o.Selector).
			SelectAllParam(o.All).
			ResourceTypeOrNameArgs(true, o.Args...)
	default:
		b = b.
			LabelSelectorParam(o.Selector).
			SelectAllParam(o.All).
			ResourceNames("imagestreams", o.Args...).
			Latest()
	}

	singleItemImplied := false
	infos, err := b.Do().IntoSingleItemImplied(&singleItemImplied).Infos()
	if err != nil {
		return err
	}

	if o.PrintTable {
		return o.printImageLookup(infos)
	}

	patches := CalculatePatchesExternal(infos, func(info *resource.Info) (bool, error) {
		switch t := info.Object.(type) {
		case *imagev1.ImageStream:
			t.Spec.LookupPolicy.Local = o.Enabled
			return true, nil
		default:
			accessor, ok := ometa.GetAnnotationAccessor(info.Object)
			if !ok {
				return true, fmt.Errorf("the resource %s does not support altering image lookup", getObjectName(info))
			}
			templateAnnotations, ok := accessor.TemplateAnnotations()
			if ok {
				if o.Enabled {
					if templateAnnotations == nil {
						templateAnnotations = make(map[string]string)
					}
					templateAnnotations[alphaResolveNamesAnnotation] = "*"
				} else {
					delete(templateAnnotations, alphaResolveNamesAnnotation)
				}
				accessor.SetTemplateAnnotations(templateAnnotations)
				return true, nil
			}
			annotations := accessor.Annotations()
			if o.Enabled {
				if annotations == nil {
					annotations = make(map[string]string)
				}
				annotations[alphaResolveNamesAnnotation] = "*"
			} else {
				delete(annotations, alphaResolveNamesAnnotation)
			}
			accessor.SetAnnotations(annotations)
			return true, nil
		}
	})

	allErrs := []error{}
	for _, patch := range patches {
		info := patch.Info
		name := getObjectName(info)
		if patch.Err != nil {
			allErrs = append(allErrs, fmt.Errorf("error: %s %v\n", name, patch.Err))
			continue
		}

		if string(patch.Patch) == "{}" || len(patch.Patch) == 0 {
			klog.V(1).Infof("info: %s was not changed\n", name)
			continue
		}

		if o.Local || o.DryRun {
			if err := o.Printer.PrintObj(info.Object, o.Out); err != nil {
				allErrs = append(allErrs, err)
			}
			continue
		}

		actual, err := o.Client.Resource(info.Mapping.Resource).Namespace(info.Namespace).Patch(info.Name, types.StrategicMergePatchType, patch.Patch, metav1.UpdateOptions{})
		if err != nil {
			allErrs = append(allErrs, fmt.Errorf("failed to patch image lookup: %v\n", err))
			continue
		}

		if err := o.Printer.PrintObj(actual, o.Out); err != nil {
			allErrs = append(allErrs, err)
		}
	}
	return utilerrors.NewAggregate(allErrs)

}
597361
func (d *StaticDiscovery) Run(looper director.Looper) {
	var err error

	d.Targets, err = d.ParseConfig(d.ConfigFile)
	if err != nil {
		log.Errorf("StaticDiscovery cannot parse: %s", err.Error())
		looper.Done(nil)
	}
}
1085065
func (c *dumpCommand) Run(ctx *cmd.Context) error {
	client, err := c.getAPI()
	if err != nil {
		return err
	}
	defer client.Close()

	_, modelDetails, err := c.ModelCommandBase.ModelDetails()
	if err != nil {
		return errors.Annotate(err, "getting model details")
	}

	modelTag := names.NewModelTag(modelDetails.ModelUUID)
	results, err := client.DumpModel(modelTag, c.simplified)
	if err != nil {
		return err
	}

	return c.out.Write(ctx, results)
}
254038
func (o *SetResourcesOptions) Run() error {
	allErrs := []error{}
	patches := CalculatePatches(o.Infos, scheme.DefaultJSONEncoder(), func(obj runtime.Object) ([]byte, error) {
		transformed := false
		_, err := o.UpdatePodSpecForObject(obj, func(spec *v1.PodSpec) error {
			containers, _ := selectContainers(spec.Containers, o.ContainerSelector)
			if len(containers) != 0 {
				for i := range containers {
					if len(o.Limits) != 0 && len(containers[i].Resources.Limits) == 0 {
						containers[i].Resources.Limits = make(v1.ResourceList)
					}
					for key, value := range o.ResourceRequirements.Limits {
						containers[i].Resources.Limits[key] = value
					}

					if len(o.Requests) != 0 && len(containers[i].Resources.Requests) == 0 {
						containers[i].Resources.Requests = make(v1.ResourceList)
					}
					for key, value := range o.ResourceRequirements.Requests {
						containers[i].Resources.Requests[key] = value
					}
					transformed = true
				}
			} else {
				allErrs = append(allErrs, fmt.Errorf("error: unable to find container named %s", o.ContainerSelector))
			}
			return nil
		})
		if err != nil {
			return nil, err
		}
		if !transformed {
			return nil, nil
		}
		// record this change (for rollout history)
		if err := o.Recorder.Record(obj); err != nil {
			klog.V(4).Infof("error recording current command: %v", err)
		}

		return runtime.Encode(scheme.DefaultJSONEncoder(), obj)
	})

	for _, patch := range patches {
		info := patch.Info
		name := info.ObjectName()
		if patch.Err != nil {
			allErrs = append(allErrs, fmt.Errorf("error: %s %v\n", name, patch.Err))
			continue
		}

		//no changes
		if string(patch.Patch) == "{}" || len(patch.Patch) == 0 {
			allErrs = append(allErrs, fmt.Errorf("info: %s was not changed\n", name))
			continue
		}

		if o.Local || o.DryRun {
			if err := o.PrintObj(info.Object, o.Out); err != nil {
				allErrs = append(allErrs, err)
			}
			continue
		}

		actual, err := resource.NewHelper(info.Client, info.Mapping).Patch(info.Namespace, info.Name, types.StrategicMergePatchType, patch.Patch, nil)
		if err != nil {
			allErrs = append(allErrs, fmt.Errorf("failed to patch limit update to pod template %v", err))
			continue
		}

		if err := o.PrintObj(actual, o.Out); err != nil {
			allErrs = append(allErrs, err)
		}
	}
	return utilerrors.NewAggregate(allErrs)
}
791881
func (u *Updater) Run(ctx context.Context, slots []orchestrator.Slot) {
	defer close(u.doneChan)

	service := u.newService

	// If the update is in a PAUSED state, we should not do anything.
	if service.UpdateStatus != nil &&
		(service.UpdateStatus.State == api.UpdateStatus_PAUSED ||
			service.UpdateStatus.State == api.UpdateStatus_ROLLBACK_PAUSED) {
		return
	}

	var dirtySlots []orchestrator.Slot
	for _, slot := range slots {
		if u.isSlotDirty(slot) {
			dirtySlots = append(dirtySlots, slot)
		}
	}
	// Abort immediately if all tasks are clean.
	if len(dirtySlots) == 0 {
		if service.UpdateStatus == nil {
			if u.annotationsUpdated(service) {
				// Annotation-only update; mark the update as completed
				u.completeUpdate(ctx, service.ID, true)
			}
			return
		}
		if service.UpdateStatus.State == api.UpdateStatus_UPDATING || service.UpdateStatus.State == api.UpdateStatus_ROLLBACK_STARTED {
			// Update or rollback was started, and is now complete
			u.completeUpdate(ctx, service.ID, true)
			return
		}
		return
	}

	// If there's no update in progress, we are starting one.
	if service.UpdateStatus == nil {
		u.startUpdate(ctx, service.ID)
	}

	var (
		monitoringPeriod time.Duration
		updateConfig     *api.UpdateConfig
	)

	if service.UpdateStatus != nil && service.UpdateStatus.State == api.UpdateStatus_ROLLBACK_STARTED {
		monitoringPeriod, _ = gogotypes.DurationFromProto(defaults.Service.Rollback.Monitor)
		updateConfig = service.Spec.Rollback
		if updateConfig == nil {
			updateConfig = defaults.Service.Rollback
		}
	} else {
		monitoringPeriod, _ = gogotypes.DurationFromProto(defaults.Service.Update.Monitor)
		updateConfig = service.Spec.Update
		if updateConfig == nil {
			updateConfig = defaults.Service.Update
		}
	}

	parallelism := int(updateConfig.Parallelism)
	if updateConfig.Monitor != nil {
		newMonitoringPeriod, err := gogotypes.DurationFromProto(updateConfig.Monitor)
		if err == nil {
			monitoringPeriod = newMonitoringPeriod
		}
	}

	if parallelism == 0 {
		// TODO(aluzzardi): We could try to optimize unlimited parallelism by performing updates in a single
		// goroutine using a batch transaction.
		parallelism = len(dirtySlots)
	}

	// Start the workers.
	slotQueue := make(chan orchestrator.Slot)
	wg := sync.WaitGroup{}
	wg.Add(parallelism)
	for i := 0; i < parallelism; i++ {
		go func() {
			u.worker(ctx, slotQueue, updateConfig)
			wg.Done()
		}()
	}

	var failedTaskWatch chan events.Event

	if updateConfig.FailureAction != api.UpdateConfig_CONTINUE {
		var cancelWatch func()
		failedTaskWatch, cancelWatch = state.Watch(
			u.store.WatchQueue(),
			api.EventUpdateTask{
				Task:   &api.Task{ServiceID: service.ID, Status: api.TaskStatus{State: api.TaskStateRunning}},
				Checks: []api.TaskCheckFunc{api.TaskCheckServiceID, state.TaskCheckStateGreaterThan},
			},
		)
		defer cancelWatch()
	}

	stopped := false
	failedTasks := make(map[string]struct{})
	totalFailures := 0

	failureTriggersAction := func(failedTask *api.Task) bool {
		// Ignore tasks we have already seen as failures.
		if _, found := failedTasks[failedTask.ID]; found {
			return false
		}

		// If this failed/completed task is one that we
		// created as part of this update, we should
		// follow the failure action.
		u.updatedTasksMu.Lock()
		startedAt, found := u.updatedTasks[failedTask.ID]
		u.updatedTasksMu.Unlock()

		if found && (startedAt.IsZero() || time.Since(startedAt) <= monitoringPeriod) {
			failedTasks[failedTask.ID] = struct{}{}
			totalFailures++
			if float32(totalFailures)/float32(len(dirtySlots)) > updateConfig.MaxFailureRatio {
				switch updateConfig.FailureAction {
				case api.UpdateConfig_PAUSE:
					stopped = true
					message := fmt.Sprintf("update paused due to failure or early termination of task %s", failedTask.ID)
					u.pauseUpdate(ctx, service.ID, message)
					return true
				case api.UpdateConfig_ROLLBACK:
					// Never roll back a rollback
					if service.UpdateStatus != nil && service.UpdateStatus.State == api.UpdateStatus_ROLLBACK_STARTED {
						message := fmt.Sprintf("rollback paused due to failure or early termination of task %s", failedTask.ID)
						u.pauseUpdate(ctx, service.ID, message)
						return true
					}
					stopped = true
					message := fmt.Sprintf("update rolled back due to failure or early termination of task %s", failedTask.ID)
					u.rollbackUpdate(ctx, service.ID, message)
					return true
				}
			}
		}

		return false
	}

slotsLoop:
	for _, slot := range dirtySlots {
	retryLoop:
		for {
			// Wait for a worker to pick up the task or abort the update, whichever comes first.
			select {
			case <-u.stopChan:
				stopped = true
				break slotsLoop
			case ev := <-failedTaskWatch:
				if failureTriggersAction(ev.(api.EventUpdateTask).Task) {
					break slotsLoop
				}
			case slotQueue <- slot:
				break retryLoop
			}
		}
	}

	close(slotQueue)
	wg.Wait()

	if !stopped {
		// Keep watching for task failures for one more monitoringPeriod,
		// before declaring the update complete.
		doneMonitoring := time.After(monitoringPeriod)
	monitorLoop:
		for {
			select {
			case <-u.stopChan:
				stopped = true
				break monitorLoop
			case <-doneMonitoring:
				break monitorLoop
			case ev := <-failedTaskWatch:
				if failureTriggersAction(ev.(api.EventUpdateTask).Task) {
					break monitorLoop
				}
			}
		}
	}

	// TODO(aaronl): Potentially roll back the service if not enough tasks
	// have reached RUNNING by this point.

	if !stopped {
		u.completeUpdate(ctx, service.ID, false)
	}
}
1068337
func (u *DropsondeUnmarshallerCollection) Run(inputChan <-chan []byte, outputChan chan<- *events.Envelope, waitGroup *sync.WaitGroup) {
	for _, unmarshaller := range u.unmarshallers {
		go func(um *DropsondeUnmarshaller) {
			defer waitGroup.Done()
			um.Run(inputChan, outputChan)
		}(unmarshaller)
	}
}
480100
func (c *poolUpdateCommand) Run(ctx *cmd.Context) (err error) {
	api, err := c.newAPIFunc()
	if err != nil {
		return err
	}
	defer api.Close()
	if api.BestAPIVersion() < 5 {
		return errors.New("updating storage pools is not supported by this version of Juju")
	}
	err = api.UpdatePool(c.poolName, c.provider, c.configAttrs)
	if err != nil {
		return err
	}
	return nil
}
807789
func (sched *Scheduler) Run() {
	if !sched.config.WaitForCacheSync() {
		return
	}

	go wait.Until(sched.scheduleOne, 0, sched.config.StopEverything)
}