Method SyncUpJobConfigs

subscriber/common/job/controller.go:277–424

SyncUpJobConfigs sync up jobConfigs with aresDB controller

()

// SyncUpJobConfigs sync up jobConfigs with aresDB controller
func (c *Controller) SyncUpJobConfigs() {
	c.Lock()
	defer c.Unlock()

	// Check if the hash of the assignment is changed or not
	updateHash, newAssignmentHash := c.updateAssignmentHash()
	if !updateHash {
		c.serviceConfig.Scope.Counter("syncUp.skipped").Inc(1)
		return
	}

	// Get assignment from aresDB controller since hash is changed
	assigned, err := c.aresControllerClient.GetAssignment(c.jobNS, c.serviceConfig.Environment.InstanceID)
	if err != nil {
		c.serviceConfig.Logger.Error("Failed to get assignment from aresDB controller",
			zap.String("jobNamespace", c.jobNS),
			zap.String("aresDB Controller", c.serviceConfig.ControllerConfig.Address),
			zap.Error(err))
		c.serviceConfig.Scope.Counter("syncUp.failed").Inc(1)
		return
	}

	assignment, err := rules.NewAssignmentFromController(assigned)
	if err != nil {
		c.serviceConfig.Logger.Error("Failed to populate assignment from controller assignment",
			zap.String("jobNamespace", c.jobNS),
			zap.String("aresDB Controller", c.serviceConfig.ControllerConfig.Address),
			zap.Error(err))
		c.serviceConfig.Scope.Counter("syncUp.failed").Inc(1)
		return
	}

	c.serviceConfig.Logger.Info("Got assignment from aresDB controller",
		zap.String("jobNamespace", c.jobNS),
		zap.String("aresDB Controller", c.serviceConfig.ControllerConfig.Address),
		zap.String("activeAresNameSpace", config.ActiveAresNameSpace),
		zap.Any("aresClusterNSConfig", c.serviceConfig.AresNSConfig),
		zap.Any("activeAresClusters", c.serviceConfig.ActiveAresClusters),
		zap.Any("assignement", assignment))

	newJobs := make(map[string]*rules.JobConfig)
	// Add or Update jobs
	for _, jobConfig := range assignment.Jobs {
		newJobs[jobConfig.Name] = jobConfig
		if aresClusterDrivers, ok := c.Drivers[jobConfig.Name]; ok {
			// case1: existing jobConfig
			for aresCluster, driver := range aresClusterDrivers {
				if _, ok := assignment.AresClusters[aresCluster]; !ok {
					// case1.1: delete the driver because aresCluster is deleted
					activeAresCluster, exist := c.serviceConfig.ActiveAresClusters[aresCluster]
					if exist && activeAresCluster.GetSinkMode() != config.Sink_Kafka {
						c.deleteDriver(driver, aresCluster, aresClusterDrivers)
						c.serviceConfig.Logger.Info("deleted driver due to the removed aresCluster",
							zap.String("job", jobConfig.Name),
							zap.String("aresCluster", aresCluster))
					}
					continue
				}