// handleReplicationEnd listens for various events that can cause the replication to end.
//
// These are:
// * The replication signals that it has finished executing.
// * The replication is canceled.
// * Errors in lock renewal (as signaled on the passed channel).
//   In this case, the error needs to be communicated up a level.
//   The passed channel may be nil, in which case this case is not checked,
//   but the others still are.
//
// When replication finishes for any of these reasons, this function is responsible for:
// * Stopping the replication (if it has not already)
// * Destroying its session (passed in to this function.
//   Passing nil is legal, in which case it is not destroyed)
func (r *replication) handleReplicationEnd(session kp.Session, renewalErrCh chan error) {
	defer func() {
		close(r.quitCh)
		close(r.errCh)
		if r.rateLimiter != nil {
			r.rateLimiter.Stop()
		}
		if session != nil {
			_ = session.Destroy()
		}
	}()

	select {
	case <-r.replicationDoneCh:
	case <-r.replicationCancelledCh:
		// If the replication is enacted, wait for it to exit
		<-r.enactedCh
	case err := <-renewalErrCh:
		// communicate the error to the caller.
		r.errCh <- replicationError{
			err:     err,
			isFatal: true,
		}
		return
	}
}

// Attempts to claim a lock. If the overrideLock is set, any existing lock holder
// will be destroyed and one more attempt will be made to acquire the lock
func (r *replication) lock(session kp.Session, lockPath string, overrideLock bool) (consulutil.Unlocker, error) {
	unlocker, err := session.Lock(lockPath)

	if _, ok := err.(consulutil.AlreadyLockedError); ok {
		holder, id, err := r.store.LockHolder(lockPath)
		if err != nil {
			return nil, util.Errorf("Lock already held for %q, could not determine holder due to error: %s", lockPath, err)
		} else if holder == "" {
			// we failed to acquire this lock, but there is no outstanding
			// holder
			// this indicates that the previous holder had a LockDelay,
			// which prevents other parties from acquiring the lock for a
			// limited time
			return nil, util.Errorf("Lock for %q is blocked due to delay by previous holder", lockPath)
		} else if overrideLock {
			err = r.store.DestroyLockHolder(id)
			if err != nil {
				return nil, util.Errorf("Unable to destroy the current lock holder (%s) for %q: %s", holder, lockPath, err)
			}

			// try acquiring the lock again, but this time don't destroy holders so we don't try forever
			return r.lock(session, lockPath, false)

		} else {
			return nil, util.Errorf("Lock for %q already held by lock %q", lockPath, holder)
		}
	}

	return unlocker, err
}

// Acquires a lock on the DS that should be used by DS farm goroutines, whose
// job it is to carry out the intent of the DS
func (s *consulStore) LockForOwnership(dsID fields.ID, session kp.Session) (consulutil.Unlocker, error) {
	lockPath, err := s.dsLockPath(dsID)
	if err != nil {
		return nil, err
	}
	return session.Lock(lockPath)
}

func (s *FakeDSStore) LockForOwnership(dsID fields.ID, session kp.Session) (consulutil.Unlocker, error) {
	lockPath, err := s.dsLockPath(dsID)
	if err != nil {
		return nil, err
	}
	s.logger.Logger.Infof("Locking daemon set on the following path: '%v'", lockPath)
	return session.Lock(lockPath)
}

// Acquires a lock on the RC for ensuring that no two rolling updates are
// created that operate on the same replication controllers.  A lock on both
// the intended "new" and "old" replication controllers should be held before
// the update is created.
func (s *consulStore) LockForUpdateCreation(rcID fields.ID, session kp.Session) (consulutil.Unlocker, error) {
	updateCreationLockPath, err := s.updateCreationLockPath(rcID)
	if err != nil {
		return nil, err
	}

	return session.Lock(updateCreationLockPath)
}

func (s *fakeStore) LockForUpdateCreation(rcID fields.ID, session kp.Session) (consulutil.Unlocker, error) {
	key := fmt.Sprintf("%s/%s", rcID, "update_creation_lock")
	return session.Lock(key)
}

func (s *fakeStore) LockForOwnership(rcID fields.ID, session kp.Session) (consulutil.Unlocker, error) {
	key := fmt.Sprintf("%s/%s", rcID, "ownership_lock")
	return session.Lock(key)
}

// Like CreateRollingUpdateFromExistingRCs except will create the new RC based
// on passed parameters, using oldRCID for the old RC. The new RC and new RU
// will be created transactionally (all or nothing)
func (s consulStore) CreateRollingUpdateFromOneExistingRCWithID(
	oldRCID rc_fields.ID,
	desiredReplicas int,
	minimumReplicas int,
	leaveOld bool,
	rollDelay time.Duration,
	newRCManifest manifest.Manifest,
	newRCNodeSelector klabels.Selector,
	newRCPodLabels klabels.Set,
	newRCLabels klabels.Set,
	rollLabels klabels.Set,
) (u roll_fields.Update, err error) {
	// There are cases where this function will create the new RC and
	// subsequently fail, in which case we need to do some cleanup.

	// cleans up new RC, might be nil if we didn't create one
	var newRCCleanup func()

	// If we had an error and the rc cleanup function is set, run it
	defer func() {
		if err != nil && newRCCleanup != nil {
			newRCCleanup()
		}
	}()

	var session kp.Session
	var renewalErrCh chan error
	session, renewalErrCh, err = s.newRUCreationSession()
	if err != nil {
		return roll_fields.Update{}, err
	}
	defer session.Destroy()

	rcIDs := rc_fields.IDs{oldRCID}
	err = s.lockRCs(rcIDs, session)
	if err != nil {
		return roll_fields.Update{}, err
	}

	err = s.checkForConflictingUpdates(rcIDs)
	if err != nil {
		return roll_fields.Update{}, err
	}

	// Now create the new RC, first checking if our session is still valid
	var newRCID rc_fields.ID
	select {
	case err = <-renewalErrCh:
		return roll_fields.Update{}, err
	default:
		rc, err := s.rcstore.Create(newRCManifest, newRCNodeSelector, newRCPodLabels)
		if err != nil {
			return roll_fields.Update{}, err
		}

		newRCCleanup = func() {
			err := s.rcstore.Delete(newRCID, false)
			if err != nil {
				s.logger.WithError(err).Errorln("Unable to cleanup RC %s after failed RU creation attempt", newRCID)
			}
		}

		newRCID = rc.ID

		// Get a lock on the new RC we just created so no parallel
		// update creations can use it
		err = s.lockRCs(rc_fields.IDs{newRCID}, session)
		if err != nil {
			return roll_fields.Update{}, err
		}
	}

	rcIDs = append(rcIDs, newRCID)
	// Check for conflicts again in case an update was created on the new
	// RC between when we created it and locked it
	err = s.checkForConflictingUpdates(rcIDs)
	if err != nil {
		return roll_fields.Update{}, err
	}

	err = s.labeler.SetLabels(labels.RC, newRCID.String(), newRCLabels)
	if err != nil {
		return roll_fields.Update{}, err
	}

	u = roll_fields.Update{
		OldRC:           oldRCID,
		NewRC:           newRCID,
		DesiredReplicas: desiredReplicas,
		MinimumReplicas: minimumReplicas,
		LeaveOld:        leaveOld,
		RollDelay:       rollDelay,
	}

	return s.attemptRUCreation(u, rollLabels, renewalErrCh)
}