if err != nil {
			bugLogIf(ctx, stream.AddError(err.Error()))
			return
		}
		err = stream.AddKeyEncrypted(clusterKey)
		if err != nil {
			bugLogIf(ctx, stream.AddError(err.Error()))
			return
		}
		if b := getClusterMetaInfo(ctx); len(b) > 0 {
			w, err := stream.AddEncryptedStream("cluster.info", nil)
			if err != nil {
				bugLogIf(ctx, err)
				return
			}

	for promMetric := range ch {
		dtoMetric := &dto.Metric{}
		err := promMetric.Write(dtoMetric)
		if err != nil {
			// Log error and continue to receive other metric
			// values
			bugLogIf(GlobalContext, err)
			continue
		}

		h := dtoMetric.GetHistogram()
		for _, b := range h.Bucket {
			labels := make(map[string]string)
			for _, lp := range dtoMetric.GetLabel() {

				atomic.AddInt32(&p.activeMRFWorkers, 1)
				replicateObject(p.ctx, v, p.objLayer)
				atomic.AddInt32(&p.activeMRFWorkers, -1)
				p.stats.decQ(v.Bucket, v.Size, v.DeleteMarker, v.OpType)

			default:
				bugLogIf(p.ctx, fmt.Errorf("unknown mrf replication type: %T", oi), "unknown-mrf-replicate-type")
			}
		case <-p.mrfWorkerKillCh:
			return
		}
	}
}

// AddWorker adds a replication worker to the pool.

				if err != nil {
					replLogIf(ctx, err)
					continue
				}
				tgtBytes, err := json.Marshal(&targets)
				if err != nil {
					bugLogIf(ctx, err)
					continue
				}
				if _, err = globalBucketMetadataSys.Update(ctx, bucket, bucketTargetsFile, tgtBytes); err != nil {
					replLogIf(ctx, err)
					continue
				}
			}
		}
	}