// markNodePhase marks a node with the given phase, creating the node if necessary and handles timestamps
513920
func (woc *wfOperationCtx) markNodePhase(nodeName string, phase wfv1.NodePhase, message ...string) *wfv1.NodeStatus {
	node := woc.getNodeByName(nodeName)
	if node == nil {
		panic(fmt.Sprintf("node %s uninitialized", nodeName))
	}
	if node.Phase != phase {
		woc.log.Infof("node %s phase %s -> %s", node, node.Phase, phase)
		node.Phase = phase
		woc.updated = true
	}
	if len(message) > 0 {
		if message[0] != node.Message {
			woc.log.Infof("node %s message: %s", node, message[0])
			node.Message = message[0]
			woc.updated = true
		}
	}
	if node.Completed() && node.FinishedAt.IsZero() {
		node.FinishedAt = metav1.Time{Time: time.Now().UTC()}
		woc.log.Infof("node %s finished: %s", node, node.FinishedAt)
		woc.updated = true
	}
	woc.wf.Status.Nodes[node.ID] = *node
	return node
}
232853
func (node *DDL) Format(buf *TrackedBuffer) {
	switch node.Action {
	case CreateStr:
		if node.OptLike != nil {
			buf.Myprintf("%s table %v %v", node.Action, node.Table, node.OptLike)
		} else if node.TableSpec != nil {
			buf.Myprintf("%s table %v %v", node.Action, node.Table, node.TableSpec)
		} else {
			buf.Myprintf("%s table %v", node.Action, node.Table)
		}
	case DropStr:
		exists := ""
		if node.IfExists {
			exists = " if exists"
		}
		buf.Myprintf("%s table%s %v", node.Action, exists, node.FromTables)
	case RenameStr:
		buf.Myprintf("%s table %v to %v", node.Action, node.FromTables[0], node.ToTables[0])
		for i := 1; i < len(node.FromTables); i++ {
			buf.Myprintf(", %v to %v", node.FromTables[i], node.ToTables[i])
		}
	case AlterStr:
		if node.PartitionSpec != nil {
			buf.Myprintf("%s table %v %v", node.Action, node.Table, node.PartitionSpec)
		} else {
			buf.Myprintf("%s table %v", node.Action, node.Table)
		}
	case FlushStr:
		buf.Myprintf("%s", node.Action)
	case CreateVindexStr:
		buf.Myprintf("alter vschema create vindex %v %v", node.VindexSpec.Name, node.VindexSpec)
	case DropVindexStr:
		buf.Myprintf("alter vschema drop vindex %v", node.VindexSpec.Name)
	case AddVschemaTableStr:
		buf.Myprintf("alter vschema add table %v", node.Table)
	case DropVschemaTableStr:
		buf.Myprintf("alter vschema drop table %v", node.Table)
	case AddColVindexStr:
		buf.Myprintf("alter vschema on %v add vindex %v (", node.Table, node.VindexSpec.Name)
		for i, col := range node.VindexCols {
			if i != 0 {
				buf.Myprintf(", %v", col)
			} else {
				buf.Myprintf("%v", col)
			}
		}
		buf.Myprintf(")")
		if node.VindexSpec.Type.String() != "" {
			buf.Myprintf(" %v", node.VindexSpec)
		}
	case DropColVindexStr:
		buf.Myprintf("alter vschema on %v drop vindex %v", node.Table, node.VindexSpec.Name)
	default:
		buf.Myprintf("%s table %v", node.Action, node.Table)
	}
}
125831
func (shardSwap *shardSchemaSwap) markStepInProgress(uiNode *workflow.Node) {
	uiNode.Message = ""
	uiNode.State = workflowpb.WorkflowState_Running
	uiNode.Display = workflow.NodeDisplayIndeterminate
	uiNode.BroadcastChanges(false /* updateChildren */)
}
377709
func (s *Node) SetNodeCreateTime(v time.Time) *Node {
	s.NodeCreateTime = &v
	return s
}
1200571
func (n NodeStatus) Completed() bool {
	return isCompletedPhase(n.Phase) || n.IsDaemoned() && n.Phase != NodePending
}
1267971
func Mark(format string, args ...interface{}) func() {
	if !debug {
		return func() {}
	}

	Trace("START "+format, args...)
	old := logger.Prefix()
	logger.SetPrefix(old + "|    ")
	return func() {
		logger.SetPrefix(old)
		Trace("END "+format, args...)
	}
}
1274414
func (facade *Facade) Phase(entities params.Entities) params.PhaseResults {
	count := len(entities.Entities)
	results := params.PhaseResults{
		Results: make([]params.PhaseResult, count),
	}
	for i, entity := range entities.Entities {
		phase, err := facade.onePhase(entity.Tag)
		results.Results[i].Phase = phase
		results.Results[i].Error = common.ServerError(err)
	}
	return results
}
626074
func (c *cluster) markDown(endpoint string) {
	c.Lock()
	defer c.Unlock()
	for _, n := range c.members {
		// step: check if this is the node and it's marked as up - The double  checking on the
		// nodes status ensures the multiple calls don't create multiple checks
		if n.status == memberStatusUp && n.endpoint == endpoint {
			n.status = memberStatusDown
			go c.healthCheckNode(n)
			break
		}
	}
}
1039361
func (ref *Consumer) MarkPartitionOffset(topic string, partition int32, offset int64, metadata string) {
	ref.Consumer.MarkPartitionOffset(topic, partition, offset, metadata)
}
651700
func (d *Dispatcher) markNodeNotReady(id string, state api.NodeStatus_State, message string) error {
	logLocal := log.G(d.ctx).WithField("method", "(*Dispatcher).markNodeNotReady")

	dctx, err := d.isRunningLocked()
	if err != nil {
		return err
	}

	// Node is down. Add it to down nodes so that we can keep
	// track of tasks assigned to the node.
	var node *api.Node
	d.store.View(func(readTx store.ReadTx) {
		node = store.GetNode(readTx, id)
		if node == nil {
			err = fmt.Errorf("could not find node %s while trying to add to down nodes store", id)
		}
	})
	if err != nil {
		return err
	}

	expireFunc := func() {
		log.G(dctx).Debugf(`worker timed-out %s in "down" state, moving all tasks to "ORPHANED" state`, id)
		if err := d.moveTasksToOrphaned(id); err != nil {
			log.G(dctx).WithError(err).Error(`failed to move all tasks to "ORPHANED" state`)
		}

		d.downNodes.Delete(id)
	}

	d.downNodes.Add(node, expireFunc)
	logLocal.Debugf("added node %s to down nodes list", node.ID)

	status := &api.NodeStatus{
		State:   state,
		Message: message,
	}

	d.nodeUpdatesLock.Lock()
	// pluck the description out of nodeUpdates. this protects against a case
	// where a node is marked ready and a description is added, but then the
	// node is immediately marked not ready. this preserves that description
	d.nodeUpdates[id] = nodeUpdate{status: status, description: d.nodeUpdates[id].description}
	numUpdates := len(d.nodeUpdates)
	d.nodeUpdatesLock.Unlock()

	if numUpdates >= maxBatchItems {
		select {
		case d.processUpdatesTrigger <- struct{}{}:
		case <-dctx.Done():
		}
	}

	if rn := d.nodes.Delete(id); rn == nil {
		return errors.Errorf("node %s is not found in local storage", id)
	}
	logLocal.Debugf("deleted node %s from node store", node.ID)

	return nil
}
144310
func (s *GroupsService) Mark(channel string, timestamp string) *GroupsMarkCall {
	var call GroupsMarkCall
	call.service = s
	call.channel = channel
	call.timestamp = timestamp
	return &call
}
623453
func NewPreflightPhase() workflow.Phase {
	return workflow.Phase{
		Name:    "preflight",
		Short:   "Run pre-flight checks",
		Long:    "Run pre-flight checks for kubeadm init.",
		Example: preflightExample,
		Run:     runPreflight,
		InheritFlags: []string{
			options.CfgPath,
			options.IgnorePreflightErrors,
		},
	}
}
501318
func (b *ByteBuf) MarkN(n int) error {
	return b.MarkIndex(b.readerIndex + n)
}
647642
func (mig *modelMigration) SetPhase(nextPhase migration.Phase) error {
	now := mig.st.clock().Now().UnixNano()

	phase, err := mig.Phase()
	if err != nil {
		return errors.Trace(err)
	}

	if nextPhase == phase {
		return nil // Already at that phase. Nothing to do.
	}
	if !phase.CanTransitionTo(nextPhase) {
		return errors.Errorf("illegal phase change: %s -> %s", phase, nextPhase)
	}

	nextDoc := mig.statusDoc
	nextDoc.Phase = nextPhase.String()
	nextDoc.PhaseChangedTime = now
	update := bson.M{
		"phase":              nextDoc.Phase,
		"phase-changed-time": now,
	}
	if nextPhase == migration.SUCCESS {
		nextDoc.SuccessTime = now
		update["success-time"] = now
	}

	ops, err := migStatusHistoryAndOps(mig.st, nextPhase, now, mig.StatusMessage())
	if err != nil {
		return errors.Trace(err)
	}

	// If the migration aborted, make the model active again.
	if nextPhase == migration.ABORTDONE {
		ops = append(ops, txn.Op{
			C:      modelsC,
			Id:     mig.doc.ModelUUID,
			Assert: txn.DocExists,
			Update: bson.M{
				"$set": bson.M{"migration-mode": MigrationModeNone},
			},
		})
	}

	// Set end timestamps and mark migration as no longer active if a
	// terminal phase is hit.
	if nextPhase.IsTerminal() {
		nextDoc.EndTime = now
		update["end-time"] = now
		ops = append(ops, txn.Op{
			C:      migrationsActiveC,
			Id:     mig.doc.ModelUUID,
			Assert: txn.DocExists,
			Remove: true,
		})
	}

	ops = append(ops, txn.Op{
		C:      migrationsStatusC,
		Id:     mig.statusDoc.Id,
		Update: bson.M{"$set": update},
		// Ensure phase hasn't changed underneath us
		Assert: bson.M{"phase": mig.statusDoc.Phase},
	})

	if err := mig.st.db().RunTransaction(ops); err == txn.ErrAborted {
		return errors.New("phase already changed")
	} else if err != nil {
		return errors.Annotate(err, "failed to update phase")
	}

	mig.statusDoc = nextDoc
	return nil
}
102121
func (p *testDataParser) MarksCompletion(line string) bool {
	return p.testEndPattern.MatchString(line)
}
778387
func (facade *Facade) Phase(uuid string) (migration.Phase, error) {
	results := params.PhaseResults{}
	err := facade.call("Phase", uuid, &results)
	if err != nil {
		return migration.UNKNOWN, errors.Trace(err)
	}
	if count := len(results.Results); count != 1 {
		return migration.UNKNOWN, countError(count)
	}
	result := results.Results[0]
	if result.Error != nil {
		return migration.UNKNOWN, errors.Trace(result.Error)
	}
	phase, ok := migration.ParsePhase(result.Phase)
	if !ok {
		err := errors.Errorf("unknown phase %q", result.Phase)
		return migration.UNKNOWN, err
	}
	return phase, nil
}
1280414
func (cache *diskBlockCacheWrapped) Mark(
	ctx context.Context, blockID kbfsblock.ID, tag string,
	cacheType DiskBlockCacheType) error {
	cache.mtx.RLock()
	defer cache.mtx.RUnlock()
	c, err := cache.getCacheLocked(cacheType)
	if err != nil {
		return err
	}
	return c.Mark(ctx, blockID, tag)
}