// Dispatch sends encoded observations to the Analyzer based on the following
// criteria:
// 1. The dispatch interval between attempts should be atleast
//    |frequency_in_hours| as specified in the Shuffler configuration.
// 2. If frequency is met, Shuffler sends |ObservationBatch| to the Analyzer for
//    each |ObservationMetadata| key if and only if:
//    - The batch contains atleast |threshold| number of Observations, and
//    - For each eligible batch, the Observations in that batch will be
//      dispatched to the Analyzer and deleted from the Shuffler, and
//    - For each batch whose Observations are not dispatched to the Analyzer
//      because the batch size is too small, the Shuffler will delete those
//      Observations from the batch whose age is at least |disposal_age_days|
//      specified in the configuration.
//
// Between between buckets, and between the batches of a single bucket, we sleep
// for |sleepDuration|.
Dispatch sends encoded observations to the Analyzer based on the following
criteria:
1. The dispatch interval between attempts should be atleast
|frequency_in_hours| as specified in the Shuffler configuration.
2. If frequency is met, Shuffler sends |ObservationBatch| to the Analyzer for
each |ObservationMetadata| key if and only if:
- The batch contains atleast |threshold| number of Observations, and
- For each eligible batch, the Observations in that batch will be
dispatched to the Analyzer and deleted from the Shuffler, and
- For each batch whose Observations are not dispatched to the Analyzer
because the batch size is too small, the Shuffler will delete those
Observations from the batch whose age is at least |disposal_age_days|
specified in the configuration.

Between between buckets, and between the batches of a single bucket, we sleep
for |sleepDuration|.
// Dispatch sends encoded observations to the Analyzer based on the following // criteria : // 1 . The dispatch interval between attempts should be atleast // |frequency_in_hours| as specified in the Shuffler configuration . // 2 . If frequency is met Shuffler sends |ObservationBatch| to the Analyzer for // each |ObservationMetadata| key if and only if : // - The batch contains atleast |threshold| number of Observations and // - For each eligible batch the Observations in that batch will be // dispatched to the Analyzer and deleted from the Shuffler and // - For each batch whose Observations are not dispatched to the Analyzer // because the batch size is too small the Shuffler will delete those // Observations from the batch whose age is at least |disposal_age_days| // specified in the configuration . // // Between between buckets and between the batches of a single bucket we sleep // for |sleepDuration| .
func (d *Dispatcher) dispatch(sleepDuration time.Duration) {
	if d.store == nil {
		panic("Store handle is nil.")
	}
	if d.config == nil {
		panic("Shuffler config is nil.")
	}
	glog.V(5).Infoln("Start dispatching ...")
	keys, err := d.store.GetKeys()
	if err != nil {
		stackdriver.LogCountMetricf(dispatchFailed, "GetKeys() failed with error: %v", err)
		return
	}
	for _, key := range keys {
		 We use the value returned from GetNumObservations() to determine whether
		 or not to dispatch a bucket. But it's important to note that this value
		 is not necessarily exactly equal to the number of Observations in the
		 Store. This is because new Observations are being added to the store and
		 the the count is being incremented asynchronously with this method and
		 non-transactionally. In particular note that it is possible that the
		 value returned from GetNumObservations() may, temporarily, be negative.
		 We do maintain the following invariant: Let n = the value returned from
		 GetNumObservations(). Then an invocation of GetObservations() by this
		 same thread immediately afterwards will find at least n Observations.
		 (The reason this invariant holds is that this is the only thread that
		 ever deletes from the store or decrements the count. All other threads
		 first add to the store, commit, and then increment the count.) This
		 allows us to use the result of GetNumObservations() for conservative
		 thresholding: We will not dispatch a bucket unless GetNumObservations()
		 returns a value at least as large as the threshold.
		bucketSize, err := d.store.GetNumObservations(key)
		glog.V(5).Infof("Bucket size from store: [%d]", bucketSize)
		if err != nil {
			stackdriver.LogCountMetricf(dispatchFailed, "GetNumObservations() failed for key: %v with error: %v", key, err)
			continue
		}
		 Compare bucket size to the configured limit.
		if uint32(bucketSize) >= d.config.GetGlobalConfig().Threshold {
			 Dispatch bucket associated with |key| and delete it after sending.
			err := d.dispatchBucket(key, sleepDuration)
			if err != nil {
				stackdriver.LogCountMetricf(dispatchFailed, "dispatchBucket() failed for key: %v with error: %v", key, err)
				continue
			}
		} else {
			 If threshold policy is not met, loop through the messages and check
			 if any messages are in the queue for more than the allowed duration
			 |disposal_age_days|. If found, discard them, otherwise queue it back
			 in the store for the next dispatch event.
			err = d.deleteOldObservations(key, storage.GetDayIndexUtc(time.Now()), d.config.GetGlobalConfig().DisposalAgeDays)
			if err != nil {
				stackdriver.LogCountMetricf(dispatchFailed, "Error in filtering Observations for key [%v]: %v", key, err)
			}
		}
		time.Sleep(sleepDuration)
	}
}
func ( d * Dispatcher ) dispatch ( sleepDuration time . Duration ) { if d . store == nil { panic ( " " ) } if d . config == nil { panic ( " " ) } glog . V ( 5 ) . Infoln ( " " ) keys , err := d . store . GetKeys ( ) if err != nil { stackdriver . LogCountMetricf ( dispatchFailed , " " , err ) return } for _ , key := range keys { bucketSize , err := d . store . GetNumObservations ( key ) glog . V ( 5 ) . Infof ( " " , bucketSize ) if err != nil { stackdriver . LogCountMetricf ( dispatchFailed , " " , key , err ) continue } if uint32 ( bucketSize ) >= d . config . GetGlobalConfig ( ) . Threshold { err := d . dispatchBucket ( key , sleepDuration ) if err != nil { stackdriver . LogCountMetricf ( dispatchFailed , " " , key , err ) continue } } else { err = d . deleteOldObservations ( key , storage . GetDayIndexUtc ( time . Now ( ) ) , d . config . GetGlobalConfig ( ) . DisposalAgeDays ) if err != nil { stackdriver . LogCountMetricf ( dispatchFailed , " " , key , err ) } } time . Sleep ( sleepDuration ) } }
Dispatch sends encoded observations to the Analyzer based on the following
criteria:
1.
Dispatch sends encoded observations to the Analyzer based on the following criteria : 1 .
// Each bucket is either dispatched or disposed based on config and if there // are errors, processing proceeds to the next bucket in the pipeline. // Fetch bucket size for each key. // // We use the value returned from GetNumObservations() to determine whether // or not to dispatch a bucket. But it's important to note that this value // is not necessarily exactly equal to the number of Observations in the // Store. This is because new Observations are being added to the store and // the the count is being incremented asynchronously with this method and // non-transactionally. In particular note that it is possible that the // value returned from GetNumObservations() may, temporarily, be negative. // We do maintain the following invariant: Let n = the value returned from // GetNumObservations(). Then an invocation of GetObservations() by this // same thread immediately afterwards will find at least n Observations. // (The reason this invariant holds is that this is the only thread that // ever deletes from the store or decrements the count. All other threads // first add to the store, commit, and then increment the count.) This // allows us to use the result of GetNumObservations() for conservative // thresholding: We will not dispatch a bucket unless GetNumObservations() // returns a value at least as large as the threshold. // Compare bucket size to the configured limit. // Dispatch bucket associated with |key| and delete it after sending. // If threshold policy is not met, loop through the messages and check // if any messages are in the queue for more than the allowed duration // |disposal_age_days|. If found, discard them, otherwise queue it back // in the store for the next dispatch event.