diff --git a/agreementbot/consumer_protocol_handler.go b/agreementbot/consumer_protocol_handler.go index 0914e657c..81f56d5ab 100644 --- a/agreementbot/consumer_protocol_handler.go +++ b/agreementbot/consumer_protocol_handler.go @@ -456,6 +456,8 @@ func (b *BaseConsumerProtocolHandler) HandlePolicyChangeForAgreement(ag persiste return true, true, true } + no_new_priority := true + // for every priority (in order highest to lowest) in the new policy with priority lower than the current wl // if it's not in the old policy, cancel choice := -1 @@ -474,9 +476,10 @@ func (b *BaseConsumerProtocolHandler) HandlePolicyChangeForAgreement(ag persiste } if currentWL := policy.GetWorkloadWithPriority(busPol.Workloads, wlUsagePriority); currentWL == nil { - // the current workload priority is no longer in the deployment policy + // the current workload priority is no longer in the deployment policy, need a service upgrade?? glog.Infof(BCPHlogstring(b.Name(), fmt.Sprintf("current workload priority %v is no longer in policy for agreement %v", wlUsagePriority, ag.CurrentAgreementId))) - return true, false, false + no_new_priority = false + //return true, false, false } else { wl = currentWL } @@ -487,7 +490,9 @@ func (b *BaseConsumerProtocolHandler) HandlePolicyChangeForAgreement(ag persiste matchingWL := policy.GetWorkloadWithPriority(oldPolicy.Workloads, choice) if matchingWL == nil || !matchingWL.IsSame(*nextPriority) { glog.Infof(BCPHlogstring(b.Name(), fmt.Sprintf("Higher priority version added or modified. Cancelling agreement %v", ag.CurrentAgreementId))) - return true, false, false + no_new_priority = false + break + //return true, false, false } nextPriority = policy.GetNextWorkloadChoice(busPol.Workloads, choice) } @@ -535,15 +540,23 @@ func (b *BaseConsumerProtocolHandler) HandlePolicyChangeForAgreement(ag persiste } } + // will return here if business policy content doesn't change + updateType := basicprotocol.MsgUpdateTypePolicyChange + newTsCs, err := policy.Create_Terms_And_Conditions(producerPol, consumerPol, wl, ag.CurrentAgreementId, b.config.AgreementBot.DefaultWorkloadPW, b.config.AgreementBot.NoDataIntervalS, basicprotocol.PROTOCOL_CURRENT_VERSION) if err != nil { glog.Errorf(BCPHlogstring(b.Name(), fmt.Sprintf("error creating new terms and conditions: %v", err))) return false, false, false } + if !no_new_priority { + //newTsCs.Workloads = busPol.Workloads + updateType = basicprotocol.MsgUpdateTypeServiceUpgrade + } + ag.LastPolicyUpdateTime = uint64(time.Now().Unix()) - b.UpdateAgreement(&ag, basicprotocol.MsgUpdateTypePolicyChange, newTsCs, cph) + b.UpdateAgreement(&ag, updateType, newTsCs, cph) return true, true, true } diff --git a/agreementbot/persistence/workload_usage.go b/agreementbot/persistence/workload_usage.go index d37aafc92..3a045ea75 100644 --- a/agreementbot/persistence/workload_usage.go +++ b/agreementbot/persistence/workload_usage.go @@ -21,6 +21,7 @@ type WorkloadUsage struct { DisableRetry bool `json:"disable_retry"` // when true, retry and retry durations are disbled which effectively disables workload rollback VerifiedDurationS int `json:"verified_durations"` // the number of seconds for successful data verification before disabling workload rollback retries ReqsNotMet bool `json:"requirements_not_met"` // this workload usage record is not at the highest priority because the device did not meet the API spec requirements at one of the higher priorities + UpgradeStategy string `json:"upgrade_strategy"` // The upgrade strategy to upgrade to this version } func (w WorkloadUsage) String() string { @@ -37,9 +38,10 @@ func (w WorkloadUsage) String() string { "DisableRetry: %v, "+ "VerifiedDurationS: %v, "+ "ReqsNotMet: %v, "+ - "Policy: %v", + "Policy: %v, "+ + "UpgradeStategy: %v", w.Id, w.DeviceId, w.PendingUpgradeTime, w.PolicyName, w.Priority, w.RetryCount, - w.RetryDurationS, w.CurrentAgreementId, w.FirstTryTime, w.LatestRetryTime, w.DisableRetry, w.VerifiedDurationS, w.ReqsNotMet, w.Policy) + w.RetryDurationS, w.CurrentAgreementId, w.FirstTryTime, w.LatestRetryTime, w.DisableRetry, w.VerifiedDurationS, w.ReqsNotMet, w.Policy, w.UpgradeStategy) } func (w WorkloadUsage) ShortString() string { @@ -55,9 +57,10 @@ func (w WorkloadUsage) ShortString() string { "LatestRetryTime: %v, "+ "DisableRetry: %v, "+ "VerifiedDurationS: %v, "+ - "ReqsNotMet: %v", + "ReqsNotMet: %v, "+ + "UpgradeStategy: %v", w.Id, w.DeviceId, w.PendingUpgradeTime, w.PolicyName, w.Priority, w.RetryCount, - w.RetryDurationS, w.CurrentAgreementId, w.FirstTryTime, w.LatestRetryTime, w.DisableRetry, w.VerifiedDurationS, w.ReqsNotMet) + w.RetryDurationS, w.CurrentAgreementId, w.FirstTryTime, w.LatestRetryTime, w.DisableRetry, w.VerifiedDurationS, w.ReqsNotMet, w.UpgradeStategy) } // private factory method for workloadusage w/out persistence safety: diff --git a/basicprotocol/protocol.go b/basicprotocol/protocol.go index 8390cc7e1..2c6328127 100644 --- a/basicprotocol/protocol.go +++ b/basicprotocol/protocol.go @@ -79,6 +79,7 @@ func NewBAgreementVerifyReply(bp *abstractprotocol.BaseProtocolMessage, exists b // receipt of a rejection. const MsgUpdateTypeSecret = "basicagreementupdatesecret" const MsgUpdateTypePolicyChange = "basicagreementtupdatepolicychange" +const MsgUpdateTypeServiceUpgrade = "basicagreementupdateServiceUpgrade" type BAgreementUpdate struct { *abstractprotocol.BaseProtocolMessage @@ -109,6 +110,10 @@ func (b *BAgreementUpdate) IsPolicyChangeUpdate() bool { return b.Updatetype == MsgUpdateTypePolicyChange } +func (b *BAgreementUpdate) IsServiceUpgrade() bool { + return b.Updatetype == MsgUpdateTypeServiceUpgrade +} + func (b *BAgreementUpdate) UpdateType() string { return b.Updatetype } @@ -148,6 +153,10 @@ func (b *BAgreementUpdateReply) IsPolicyChangeUpdate() bool { return b.Updatetype == MsgUpdateTypePolicyChange } +func (b *BAgreementUpdateReply) IsServiceUpgrade() bool { + return b.Updatetype == MsgUpdateTypeServiceUpgrade +} + func (b *BAgreementUpdateReply) IsAccepted() bool { return b.Accepted } diff --git a/businesspolicy/business_policy.go b/businesspolicy/business_policy.go index df30234c8..58d473b68 100644 --- a/businesspolicy/business_policy.go +++ b/businesspolicy/business_policy.go @@ -95,12 +95,14 @@ func (w WorkloadPriority) String() string { type UpgradePolicy struct { Lifecycle string `json:"lifecycle,omitempty"` // immediate, never, agreement Time string `json:"time,omitempty"` // the time of the upgrade + Strategy string `json:"strategy,omitempty"` // rolling (start new, stop old), recreate (stop old, start new). Default is rolling. } func (w UpgradePolicy) String() string { - return fmt.Sprintf("Lifecycle: %v, Time: %v", + return fmt.Sprintf("Lifecycle: %v, Time: %v, Strategy: %v", w.Lifecycle, - w.Time) + w.Time, + w.Strategy) } type WorkloadChoice struct { diff --git a/events/events.go b/events/events.go index eef9d253a..732dca825 100644 --- a/events/events.go +++ b/events/events.go @@ -97,6 +97,7 @@ const ( // Service related SERVICE_CONFIG_STATE_CHANGED EventId = "SERVICE_CONFIG_STATE_CHANGED" + SERVICE_VERSION_UPGRADE EventId = "SERVICE_VERSION_UPGRADE" // Object Policy related OBJECT_POLICY_NEW EventId = "OBJECT_POLICY_NEW" @@ -2216,6 +2217,45 @@ func NewWorkloadUpdateMessage(id EventId, agreementId string, protocol string, c } } +type WorkloadUpgradeMessage struct { + event Event + launchContext *AgreementLaunchContext + // AgreementProtocol string + // AgreementId string + // Deployment persistence.DeploymentConfig + // WorkloadUpdate policy.Workload +} + +func (w *WorkloadUpgradeMessage) Event() Event { + return w.event +} + +func (w *WorkloadUpgradeMessage) String() string { + //depStr := "" + // if w.Deployment != nil { + // depStr = w.Deployment.ToString() + // } + + //return fmt.Sprintf("event: %v, AgreementProtocol: %v, agreementId: %v, deployment: %v, workloadUpdate: %v", w.event, w.AgreementProtocol, w.AgreementId, depStr, w.WorkloadUpdate) + return "" +} + +func (w *WorkloadUpgradeMessage) ShortString() string { + return w.String() +} + +func NewWorkloadUpgradeMessage(id EventId, agreementId string, protocol string, deployment persistence.DeploymentConfig, workloadUpdate policy.Workload) *WorkloadUpgradeMessage { + return &WorkloadUpgradeMessage{ + event: Event{ + Id: id, + }, + // AgreementId: agreementId, + // AgreementProtocol: protocol, + // Deployment: deployment, + // WorkloadUpdate: workloadUpdate, + } +} + type NMPStartDownloadMessage struct { event Event Message StartDownloadMessage diff --git a/governance/governance.go b/governance/governance.go index b074d7d34..f8451802e 100644 --- a/governance/governance.go +++ b/governance/governance.go @@ -893,7 +893,7 @@ func (w *GovernanceWorker) CommandHandler(command worker.Command) bool { // See if it is a proposal first. If it is, ignore it... If not, check if message still exists and then check for the message types this modules cares about if _, err := protocolHandler.ValidateProposal(protocolMsg); err == nil { // Gets in here if it is a proposal - glog.V(5).Infof(logString(fmt.Sprintf("Governance handler ignoring proposal message"))) + glog.V(5).Infof(logString(fmt.Sprintf("Governance handler ignoring proposal message"))) //??? } else if there, err := w.messageInExchange(exchangeMsg.MsgId); err != nil { glog.Errorf(logString(fmt.Sprintf("unable to get messages from the exchange, error %v", err))) w.AddDeferredCommand(cmd) @@ -1113,7 +1113,7 @@ func (w *GovernanceWorker) CommandHandler(command worker.Command) bool { } else { // Allow the message extension handler to see the message - handled, cancel, agid, updatedSecs, err := w.producerPH[msgProtocol].HandleExtensionMessages(&cmd.Msg, exchangeMsg) + handled, cancel, agid, updatedSecs, updatedWorkload, err := w.producerPH[msgProtocol].HandleExtensionMessages(&cmd.Msg, exchangeMsg) if err != nil { glog.Errorf(logString(fmt.Sprintf("unable to handle message %v , error: %v", protocolMsg, err))) } else if cancel { @@ -1167,11 +1167,102 @@ func (w *GovernanceWorker) CommandHandler(command worker.Command) bool { if len(updatedSecs) != 0 { clusterNamespaceInAg, err := w.GetRequestedClusterNamespaceFromAg(&ags[0]) if err != nil { - glog.Errorf(logString(fmt.Sprintf("Failed to get cluster namespace from agreeent %v. %v", ags[0].CurrentAgreementId, err))) + glog.Errorf(logString(fmt.Sprintf("Failed to get cluster namespace from agreement %v. %v", ags[0].CurrentAgreementId, err))) } // have updatedSecs, send out an event to let kube worker know about the secret update w.Messages() <- events.NewWorkloadUpdateMessage(events.UPDATE_SECRETS_IN_AGREEMENT, agid, msgProtocol, clusterNamespaceInAg, ags[0].GetDeploymentConfig(), updatedSecs) } + + if updatedWorkload != nil { + // need to set up agreementLaunchContext() + clusterNamespaceInAg, err := w.GetRequestedClusterNamespaceFromAg(&ags[0]) + if err != nil { + glog.Errorf(logString(fmt.Sprintf("Failed to get cluster namespace from agreement %v. %v", ags[0].CurrentAgreementId, err))) + } + + // get service image auths from the exchange + img_auths, err := w.GetImageAuths(updatedWorkload) + if err != nil { + glog.Errorf(logString(fmt.Sprintf("Failed to get image auth for workload %v in agreement. %v", updatedWorkload, ags[0].CurrentAgreementId, err))) + } + + cc := events.NewContainerConfig(updatedWorkload.Deployment, updatedWorkload.DeploymentSignature, updatedWorkload.DeploymentUserInfo, + updatedWorkload.ClusterDeployment, updatedWorkload.ClusterDeploymentSignature, clusterNamespaceInAg, updatedWorkload.DeploymentOverrides, img_auths) + + lc := new(events.AgreementLaunchContext) + lc.Configure = *cc + lc.AgreementId = ags[0].CurrentAgreementId + lc.AgreementProtocol = ags[0].AgreementProtocol + + // get environmental settings for the workload + + // The service config variables are stored in the device's attributes. + if proposal, err := protocolHandler.DemarshalProposal(ags[0].Proposal); err != nil { + + } else if tcPolicy, err := policy.DemarshalPolicy(proposal.TsAndCs()); err != nil { + + } else { + envAdds, err := w.GetServicePreference(updatedWorkload.WorkloadURL, updatedWorkload.Org, tcPolicy) + if err != nil { + glog.Errorf(logString(fmt.Sprintf("Error getting environment variables from node settings for %v %v: %v", updatedWorkload.WorkloadURL, updatedWorkload.Org, err))) + } + + // The workload config we have might be from a lower version of the workload. Go to the exchange and + // get the metadata for the version we are running and then add in any unset default user inputs. + var serviceDef *exchange.ServiceDefinition + serviceId := "" + if _, sDef, allIDs, err := exchange.GetHTTPServiceResolverHandler(w)(updatedWorkload.WorkloadURL, updatedWorkload.Org, updatedWorkload.Version, updatedWorkload.Arch); err != nil { + //return fmt.Errorf(logString(fmt.Sprintf("Received error querying exchange for service metadata: %v/%v, error %v", workload.Org, workload.WorkloadURL, err))) + } else if sDef == nil { + //return fmt.Errorf(logString(fmt.Sprintf("Cound not find service metadata for %v/%v.", workload.Org, workload.WorkloadURL))) + } else { + serviceDef = sDef + sDef.PopulateDefaultUserInput(envAdds) + if allIDs != nil && len(allIDs) > 0 { + serviceId = allIDs[0] + } + } + + // create microservice def for this agreement + var msdef *persistence.MicroserviceDefinition + msFilters := []persistence.MSFilter{persistence.UrlOrgVersionMSFilter(serviceDef.URL, exchange.GetOrg(serviceId), serviceDef.Version), persistence.UnarchivedMSFilter()} + if msdefs, err := persistence.FindMicroserviceDefs(w.db, msFilters); err != nil { + //return fmt.Errorf(logString(fmt.Sprintf("Error finding service definition from the local db for %v. %v", serviceId, err))) + } else if msdefs == nil || len(msdefs) == 0 { + vExp, err1 := semanticversion.Version_Expression_Factory(serviceDef.Version) + if err1 != nil { + //return fmt.Errorf("VersionRange %v cannot be converted to a version expression, error %v", serviceDef.Version, err1) + } + + if msdef, err = microservice.CreateMicroserviceDefWithServiceDef(w.db, serviceDef, serviceId, vExp.Get_expression()); err != nil { + //return fmt.Errorf(logString(fmt.Sprintf("failed to create service definition for %v for agreement %v: %v", serviceId, proposal.AgreementId(), err))) + } + } else { + msdef = &msdefs[0] + } + + cutil.SetPlatformEnvvars(envAdds, + config.ENVVAR_PREFIX, + proposal.AgreementId(), + exchange.GetId(w.GetExchangeId()), + exchange.GetOrg(w.GetExchangeId()), + w.GetExchangeURL(), + w.devicePattern, + w.BaseWorker.Manager.Config.GetFileSyncServiceProtocol(), + w.BaseWorker.Manager.Config.GetFileSyncServiceAPIListen(), + strconv.Itoa(int(w.BaseWorker.Manager.Config.GetFileSyncServiceAPIPort()))) + + lc.EnvironmentAdditions = &envAdds + + if err := w.processServiceSecrets(tcPolicy, proposal.AgreementId()); err != nil { + + } + + w.Messages() <- events.NewWorkloadUpgradeMessage(events.SERVICE_VERSION_UPGRADE, agid, msgProtocol, ags[0].GetDeploymentConfig(), *updatedWorkload) + + } + + } } } @@ -2144,3 +2235,26 @@ func (w *GovernanceWorker) GetRequestedClusterNamespaceFromAg(ag *persistence.Es return tcPolicy.ClusterNamespace, nil } } + +func (w *GovernanceWorker) GetImageAuths(workload *policy.Workload) ([]events.ImageDockerAuth, error) { + img_auths := make([]events.ImageDockerAuth, 0) + if w.deviceType == persistence.DEVICE_TYPE_DEVICE { + if w.Config.Edge.TrustDockerAuthFromOrg { + if ias, err := exchange.GetHTTPServiceDockerAuthsHandler(w)(workload.WorkloadURL, workload.Org, workload.Version, workload.Arch); err != nil { + return img_auths, errors.New(logString(fmt.Sprintf("received error querying exchange for service image auths: %v, error %v", workload, err))) + } else { + if ias != nil { + for _, iau_temp := range ias { + username := iau_temp.UserName + if username == "" { + username = "token" + } + img_auths = append(img_auths, events.ImageDockerAuth{Registry: iau_temp.Registry, UserName: username, Password: iau_temp.Token}) + } + } + } + } + } + + return img_auths, nil +} diff --git a/governance/microservice.go b/governance/microservice.go index 3a8788270..b848a1e3b 100644 --- a/governance/microservice.go +++ b/governance/microservice.go @@ -847,8 +847,21 @@ func (w *GovernanceWorker) handleMicroserviceUpgrade(msdef_id string) { persistence.NewMessageMeta(EL_GOV_ERR_RETRIEVE_SDEFS_FROM_DB, msdef_id, err.Error()), persistence.EC_DATABASE_ERROR) glog.Errorf(logString(fmt.Sprintf("error getting service definitions %v from db. %v", msdef_id, err))) - } else if microservice.MicroserviceReadyForUpgrade(msdef, w.db) { - // find the new ms def to upgrade to + } else { + + } + + + else if microservice.MicroserviceReadyForUpgrade(msdef, w.db) { // this is only for dependent service upgrade + // case 1: dependent service: call microservice.MicroserviceReadyForUpgrade(msdef, w.db): find the new ms def to upgrade to + + // case 2: top level service: get the version need to be upgrade to from bp? + // 1. get 1) upgrade policy 2) version to upgrade from bp or from microserviceDef + // 2. call UpgradeMicroservice() + // 1) stop old, start new -> like today + // 2) start new, record start time. subworker keep checking runningDuration = currentTime - startTime for all microserviceDefs which upgradeStarting is true. Once runningDuration > new_service_up_durations, stop old service + // 3) Once finish upgrade, set upgradeStarting to false + if new_msdef, err := microservice.GetUpgradeMicroserviceDef(exchange.GetHTTPServiceResolverHandler(w.limitedRetryEC), msdef, w.db); err != nil { glog.Errorf(logString(fmt.Sprintf("Error finding the new service definition to upgrade to for %v/%v version %v. %v", msdef.Org, msdef.SpecRef, msdef.Version, err))) } else if new_msdef == nil { diff --git a/imagefetch/image_process.go b/imagefetch/image_process.go index 073a8e2f2..3adee11d8 100644 --- a/imagefetch/image_process.go +++ b/imagefetch/image_process.go @@ -68,6 +68,11 @@ func (w *ImageFetchWorker) NewEvent(incoming events.Message) { fCmd := w.NewFetchCommand(msg.LaunchContext()) w.Commands <- fCmd + case *events.WorkloadUpgradeMessage: + msg, _ := incoming.(*events.WorkloadUpgradeMessage) + + fCmd := w.NewFetchCommand(msg.LaunchContext()) + case *events.LoadContainerMessage: msg, _ := incoming.(*events.LoadContainerMessage) diff --git a/microservice/microservice.go b/microservice/microservice.go index 7b85315f8..e835f2117 100644 --- a/microservice/microservice.go +++ b/microservice/microservice.go @@ -175,7 +175,7 @@ func MicroserviceReadyForUpgrade(msdef *persistence.MicroserviceDefinition, db * if msi.IsAgreementLess() { return false // never upgrade top level services, they are controlled by the agbot. - } else if msi.IsTopLevelService() { + } else if msi.IsTopLevelService() { // Need to remove this check return false } else if !msdef.ActiveUpgrade && msi.GetServiceDefId() == msdef.Id && msi.GetCleanupStartTime() == 0 { // If the service can only be upgraded when there are no agreements, check for agreements. diff --git a/persistence/service_upgrade.go b/persistence/service_upgrade.go new file mode 100644 index 000000000..b333ce4c1 --- /dev/null +++ b/persistence/service_upgrade.go @@ -0,0 +1,43 @@ +package persistence + +import "github.com/open-horizon/anax/policy" + +const SERVICE_UPGRADE_STATUS = "service_upgrade_status" + +// this table is to store the top level service upgrade status +// Delete record from service_upgrade_status table once the service upgrade is done +// Key is AgreementId. So for the same agreement Id, there can only have 1 upgrade at one time. +// If service upgrading (-> 2.0.0) is in progress, then agent receives agreement update to upgrade to 3.0.0, what should we do? (like nmp) + +type AgreementWorkload struct { + Workload WorkloadInfo `json:"workload"` + //Workload policy.Workload `json:"workload"` + WorkloadStartTime uint64 `json:"workload_start_time"` +} + +type ServiceUpgradeStatus struct { + AgreementId string `json:"agreement_id"` + CurrentMicroserviceDefId string `json:"current_microservice_def_id"` + CurrentRunningWorkload AgreementWorkload `json:"current_running_workload"` + UpgradeMicroserviceDefId string `json:"upgrade_microservice_def_id"` + UpgradeWorkload AgreementWorkload `json:"upgrade_workload"` + UpgradePolicy policy.UpgradePolicy `json:"upgrade_policy"` +} + +// ======== need to decide use AgreementWorkload v1 or v2 ========= + +type AgreementWorkload2 struct { + //Workload WorkloadInfo `json:"workload"` + Workload policy.Workload `json:"workload"` + WorkloadStartTime uint64 `json:"workload_start_time"` +} + +type ServiceUpgradeStatus2 struct { + AgreementId string `json:"agreement_id"` + CurrentMicroserviceDefId string `json:"current_microservice_def_id"` + CurrentRunningWorkload AgreementWorkload2 `json:"current_running_workload"` + UpgradeMicroserviceDefId string `json:"upgrade_microservice_def_id"` + UpgradeWorkload AgreementWorkload2 `json:"upgrade_workload"` + WorkloadUpgradePolicy policy.UpgradePolicy `json:"workload_upgrade_policy"` + UpgradeStatus string `json:"workload_upgrade_status"` // do we need this? +} diff --git a/policy/workload.go b/policy/workload.go index 3fd0b1ed3..9ad0b3e0b 100644 --- a/policy/workload.go +++ b/policy/workload.go @@ -61,7 +61,8 @@ type Workload struct { WorkloadPassword string `json:"workload_password,omitempty"` // The password used to create the bcrypt hash that is passed to the workload so that the workload can verify the caller ClusterDeployment string `json:"cluster_deployment,omitempty"` ClusterDeploymentSignature string `json:"cluster_deployment_signature,omitempty"` - Priority WorkloadPriority `json:"priority,omitempty"` // The highest priority workload is tried first for an agrement, if it fails, the next priority is tried. Priority 1 is the highest, priority 2 is next, etc. + Priority WorkloadPriority `json:"priority,omitempty"` // The highest priority workload is tried first for an agrement, if it fails, the next priority is tried. Priority 1 is the highest, priority 2 is next, etc. + UpgradePolicy UpgradePolicy `json:"upgrade_policy,omitempty"` WorkloadURL string `json:"workloadUrl,omitempty"` // Added with MS split, refers to a workload definition in the exchange Org string `json:"organization,omitempty"` // Added woth org support, refers to the organization where the workload is defined Version string `json:"version,omitempty"` // Added with MS split, refers to the version of the workload @@ -229,8 +230,8 @@ func GetNextWorkloadChoice(wls WorkloadList, currentChoice int) *Workload { var bestChoice Workload for _, wl := range wls { if (currentChoice == -1 || wl.Priority.PriorityValue < currentChoice) && (bestChoicePriority == -1 || wl.Priority.PriorityValue < bestChoicePriority) { - bestChoicePriority = wl.Priority.PriorityValue - bestChoice = wl + bestChoicePriority = wl.Priority.PriorityValue // 1 + bestChoice = wl // 6.0.0 } } if bestChoicePriority == -1 { @@ -247,3 +248,10 @@ func GetWorkloadWithPriority(wls WorkloadList, priority int) *Workload { } return nil } + +type UpgradePolicy struct { + Lifecycle string `json:"lifecycle,omitempty"` // immediate, never, agreement + Time string `json:"time,omitempty"` // the time of the upgrade + Strategy string `json:"strategy,omitempty"` // rolling (start new, stop old), recreate (stop old, start new). Default is rolling. + NewServiceUpDurationS int `json:"new_service_up_durations"` // if strategy is rolling. Use this field to indicate number of seconds for the new version to run before tearing down the old version +} diff --git a/producer/basic_protocol_handler.go b/producer/basic_protocol_handler.go index 341cbf01e..c064ca864 100644 --- a/producer/basic_protocol_handler.go +++ b/producer/basic_protocol_handler.go @@ -107,13 +107,13 @@ func (c *BasicProtocolHandler) VerifyAgreement(ag *persistence.EstablishedAgreem } // Returns 2 booleans, first is whether or not the message was handled, the second is whether or not to cancel the agreement in the protocol msg. -func (c *BasicProtocolHandler) HandleExtensionMessages(msg *events.ExchangeDeviceMessage, exchangeMsg *exchange.DeviceMessage) (bool, bool, string, []persistence.PersistedServiceSecret, error) { +func (c *BasicProtocolHandler) HandleExtensionMessages(msg *events.ExchangeDeviceMessage, exchangeMsg *exchange.DeviceMessage) (bool, bool, string, []persistence.PersistedServiceSecret, *policy.Workload, error) { updatedSecs := []persistence.PersistedServiceSecret{} // The agreement verification reply indicates whether or not the consumer thinks the agreement is still valid. if verify, err := c.agreementPH.ValidateAgreementVerifyReply(msg.ProtocolMessage()); err == nil { glog.V(5).Infof(BPHlogString(fmt.Sprintf("extension handler handled agreement verification reply for %v", verify.AgreementId()))) - return true, !verify.Exists, verify.AgreementId(), updatedSecs, nil + return true, !verify.Exists, verify.AgreementId(), updatedSecs, nil, nil } else if verify, err := c.agreementPH.ValidateAgreementVerify(msg.ProtocolMessage()); err == nil { // This is a request to verify that an agreement exists. @@ -139,13 +139,14 @@ func (c *BasicProtocolHandler) HandleExtensionMessages(msg *events.ExchangeDevic } } - return true, false, verify.AgreementId(), updatedSecs, nil + return true, false, verify.AgreementId(), updatedSecs, nil, nil } else if update, err := c.agreementPH.ValidateUpdate(msg.ProtocolMessage()); err == nil { // If there are no errors and the update type is accepted, send a positive reply. acceptedUpdate := true sendReply := true + var updated_workload *policy.Workload if update.IsSecretUpdate() { @@ -175,14 +176,15 @@ func (c *BasicProtocolHandler) HandleExtensionMessages(msg *events.ExchangeDevic } - } else if update.IsPolicyChangeUpdate() { + } else if update.IsPolicyChangeUpdate() || update.IsServiceUpgrade() { // a policy that was used to form an agreement with this node has changed // the agbot has sent an updated merged policy // this will update the agreement's ts&cs with the new merged policy - glog.V(5).Infof(BPHlogString(fmt.Sprintf("handling update for %v: %v", update.AgreementId(), update.Metadata))) + glog.V(5).Infof(BPHlogString(fmt.Sprintf("handling %v update for %v: %v", update.UpdateType(), update.AgreementId(), update.Metadata))) // convert the meta data into a new policy var newPolicy policy.Policy + if bytes, err := json.Marshal(update.Metadata); err != nil { glog.Errorf(BPHlogString(fmt.Sprintf("unable to marshal update for agreement %v: %v", update.AgreementId(), err))) acceptedUpdate = false @@ -212,6 +214,36 @@ func (c *BasicProtocolHandler) HandleExtensionMessages(msg *events.ExchangeDevic acceptedUpdate = false } else { glog.V(5).Infof(BPHlogString(fmt.Sprintf("updated agreement %v to %v", update.AgreementId(), updatedAg))) + if update.IsServiceUpgrade() { + // handle service upgrade + + updated_workload = newPolicy.NextHighestPriorityWorkload(0, 0, 0) + + /* + // for device case: + // get service image auths from the exchange + img_auths := make([]events.ImageDockerAuth, 0) + if w.deviceType == persistence.DEVICE_TYPE_DEVICE { + if w.Config.Edge.TrustDockerAuthFromOrg { + if ias, err := exchange.GetHTTPServiceDockerAuthsHandler(w)(workload.WorkloadURL, workload.Org, workload.Version, workload.Arch); err != nil { + return errors.New(logString(fmt.Sprintf("received error querying exchange for service image auths: %v, error %v", workload, err))) + } else { + if ias != nil { + for _, iau_temp := range ias { + username := iau_temp.UserName + if username == "" { + username = "token" + } + img_auths = append(img_auths, events.ImageDockerAuth{Registry: iau_temp.Registry, UserName: username, Password: iau_temp.Token}) + } + } + } + } + } + + cc := events.NewContainerConfig(workload.Deployment, workload.DeploymentSignature, workload.DeploymentUserInfo, + workload.ClusterDeployment, workload.ClusterDeploymentSignature, newPolicy.ClusterNamespace, workload.DeploymentOverrides, img_auths) */ + } } } else { // The update type is unexpected so simply reject it. @@ -229,11 +261,11 @@ func (c *BasicProtocolHandler) HandleExtensionMessages(msg *events.ExchangeDevic } } - return true, false, update.AgreementId(), updatedSecs, nil + return true, false, update.AgreementId(), updatedSecs, updated_workload, nil } else if reply, err := c.agreementPH.ValidateUpdateReply(msg.ProtocolMessage()); err == nil { glog.Infof(BPHlogString(fmt.Sprintf("nothing to do for update reply %v", reply))) - return true, false, update.AgreementId(), updatedSecs, nil + return true, false, update.AgreementId(), updatedSecs, nil, nil } else { @@ -241,10 +273,10 @@ func (c *BasicProtocolHandler) HandleExtensionMessages(msg *events.ExchangeDevic // so make sure it's not one of those, then we know if it's an unknown msg or not. if _, err := c.agreementPH.ValidateProposal(msg.ProtocolMessage()); err == nil { glog.V(5).Infof(BPHlogString(fmt.Sprintf("extension message handler ignoring message: %s because it is a proposal.", msg.ShortProtocolMessage()))) - return false, false, "", updatedSecs, nil + return false, false, "", updatedSecs, nil, nil } else { glog.V(3).Infof(BPHlogString(fmt.Sprintf("extension message handler ignoring message: %s because it is not a known protocol msg.", msg.ShortProtocolMessage()))) - return true, false, "", updatedSecs, nil + return true, false, "", updatedSecs, nil, nil } } diff --git a/producer/producer_protocol_handler.go b/producer/producer_protocol_handler.go index be1d389c9..a1e26f3e8 100644 --- a/producer/producer_protocol_handler.go +++ b/producer/producer_protocol_handler.go @@ -72,7 +72,7 @@ type ProducerProtocolHandler interface { SetBlockchainWritable(cmd *BCWritableCommand) IsBlockchainWritable(agreement *persistence.EstablishedAgreement) bool IsAgreementVerifiable(agreement *persistence.EstablishedAgreement) bool - HandleExtensionMessages(msg *events.ExchangeDeviceMessage, exchangeMsg *exchange.DeviceMessage) (bool, bool, string, []persistence.PersistedServiceSecret, error) + HandleExtensionMessages(msg *events.ExchangeDeviceMessage, exchangeMsg *exchange.DeviceMessage) (bool, bool, string, []persistence.PersistedServiceSecret, *policy.Workload, error) UpdateConsumer(ag *persistence.EstablishedAgreement) UpdateConsumers() GetKnownBlockchain(ag *persistence.EstablishedAgreement) (string, string, string)