kubegres 源码解析(四)

func (r *PrimaryToReplicaFailOver) promoteReplicaToPrimary(newPrimary statefulset.StatefulSetWrapper) error {
    newPrimary.StatefulSet.Labels["replicationRole"] = ctx.PrimaryRoleName
    newPrimary.StatefulSet.Spec.Template.Labels["replicationRole"] = ctx.PrimaryRoleName
    volumeMount := core.VolumeMount{
        Name:      "base-config",
        MountPath: "/tmp/promote_replica_to_primary.sh",
        SubPath:   "promote_replica_to_primary.sh",
    }

    initContainer := &newPrimary.StatefulSet.Spec.Template.Spec.InitContainers[0]
    initContainer.VolumeMounts = append(initContainer.VolumeMounts, volumeMount)
    initContainer.Command = []string{"sh", "-c", "/tmp/promote_replica_to_primary.sh"}

    err := r.activateOperationFailingOver(newPrimary)
    if err != nil {
        r.kubegresContext.Log.ErrorEvent("FailOverOperationActivationErr", err,
            "Error while activating a blocking operation for the FailOver of a Primary DB.",
            "InstanceIndex", newPrimary.InstanceIndex)
        return err
    }

    r.kubegresContext.Log.InfoEvent("FailOver", "FailOver: Promoting Replica to Primary.",
        "Replica to promote", newPrimary.StatefulSet.Name)

    err2 := r.kubegresContext.Client.Update(r.kubegresContext.Ctx, &newPrimary.StatefulSet)
    if err2 != nil {
        r.kubegresContext.Log.ErrorEvent("FailOverErr", err2,
            "FailOver: Unable to promote Replica to Primary.",
            "Replica to promote", newPrimary.StatefulSet.Name)
        r.blockingOperation.RemoveActiveOperation()
        return err
    }

    return nil
}

func (r *ReplicaDbCountSpecEnforcer) Enforce() error {
    if r.blockingOperation.IsActiveOperationIdDifferentOf(operation.OperationIdReplicaDbCountSpecEnforcement) {
        return nil
    }

    if r.hasLastAttemptTimedOut() {
        if r.isPreviouslyFailedAttemptOnReplicaDbFixed() {
            r.blockingOperation.RemoveActiveOperation()
            r.logKubegresFeaturesAreReEnabled()
        } else {
            r.logTimedOut()
            return nil
        }
    }

    if !r.isPrimaryDbReady() {
        return nil
    }

    isManualFailoverRequested := r.isManualFailoverRequested()
    if isManualFailoverRequested {
        r.resetInSpecManualFailover()
    }

    if r.isReplicaOperationInProgress() {
        return nil
    }

    // Check if the number of deployed replicas == spec if not then deploy one
    nbreNewReplicaToDeploy := r.getExpectedNbreReplicasToDeploy() - r.getNbreDeployedReplicas()

    if nbreNewReplicaToDeploy > 0 {
        if r.isAutomaticFailoverDisabled() &&
            !isManualFailoverRequested &&
            !r.doesSpecRequireTheDeploymentOfAdditionalReplicas() {
            r.logAutomaticFailoverIsDisabled()
            return nil
        }
        return r.deployReplicaStatefulSet()
    } else if nbreNewReplicaToDeploy < 0 {
        replicaToUndeploy := r.getReplicaToUndeploy()
        return r.undeployReplicaStatefulSets(replicaToUndeploy)
    } else if nbreNewReplicaToDeploy == 0 {
        for _, replicaStatefulSet := range r.getDeployedReplicas() {
            if !replicaStatefulSet.IsReady {
                return r.undeployReplicaStatefulSets(replicaStatefulSet)
            }
        }
    }

    return nil
}