resultCh := make(chan TargetReplicationResyncStatus, 1)
	defer xioutil.SafeClose(resultCh)
	go func() {
		for r := range resultCh {
			s.incStats(r, opts)
			globalSiteResyncMetrics.updateMetric(r, opts.resyncID)
		}
	}()

	var wg sync.WaitGroup
	for i := 0; i < resyncParallelRoutines; i++ {
		wg.Add(1)
		workers[i] = make(chan ReplicateObjectInfo, 100)
		i := i

	dek, err := k.conn.GenerateKey(ctx, req)
	k.updateMetrics(err, time.Since(start))

	return dek, err
}

// Decrypt decrypts a ciphertext using the master key req.Name.
// It returns ErrKeyNotFound if the key does not exist.
func (k *KMS) Decrypt(ctx context.Context, req *DecryptRequest) ([]byte, error) {
	start := time.Now()
	plaintext, err := k.conn.Decrypt(ctx, req)
	k.updateMetrics(err, time.Since(start))

}

func (pm *basicManager) UpdatePod(pod *v1.Pod) {
	pm.lock.Lock()
	defer pm.lock.Unlock()
	pm.updatePodsInternal(pod)
}

// updateMetrics updates the metrics surfaced by the pod manager.
// oldPod or newPod may be nil to signify creation or deletion.
func updateMetrics(oldPod, newPod *v1.Pod) {
	var numEC int
	if oldPod != nil {
		numEC -= len(oldPod.Spec.EphemeralContainers)
	}
	if newPod != nil {

	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			push := s.globalPushContext()
			model.LastPushMutex.Lock()
			if model.LastPushStatus != push {
				model.LastPushStatus = push
				push.UpdateMetrics()
				out, _ := model.LastPushStatus.StatusJSON()
				if string(out) != "{}" {
					log.Infof("Push Status: %s", string(out))
				}
			}
			model.LastPushMutex.Unlock()
		case <-stopCh:
			return

func (ps *PushContext) OnConfigChange() {
	LastPushMutex.Lock()
	LastPushStatus = ps
	LastPushMutex.Unlock()
	ps.UpdateMetrics()
}

// UpdateMetrics will update the prometheus metrics based on the
// current status of the push.
func (ps *PushContext) UpdateMetrics() {
	ps.proxyStatusMutex.RLock()
	defer ps.proxyStatusMutex.RUnlock()

	for _, pm := range metrics {