// dispatchBucket dispatches the ObservationBatch associated with |key| in
// chunks of size |batchSize| to Analyzer using grpc transport.
//
// We sleep for |sleepDuration| between batches.
dispatchBucket dispatches the ObservationBatch associated with |key| in
chunks of size |batchSize| to Analyzer using grpc transport.

We sleep for |sleepDuration| between batches.
// dispatchBucket dispatches the ObservationBatch associated with |key| in // chunks of size |batchSize| to Analyzer using grpc transport . // // We sleep for |sleepDuration| between batches .
func (d *Dispatcher) dispatchBucket(key *cobalt.ObservationMetadata, sleepDuration time.Duration) error {
	if key == nil {
		panic("key is nil")
	}
	if d == nil {
		panic("dispatcher is nil")
	}
	iterator, err := d.store.GetObservations(key)
	if err != nil {
		stackdriver.LogCountMetricf(dispatchBucketFailed, "GetObservations() failed for key: %v with error: %v", key, err)
		return err
	}
	batchID := 0
	for {
		batchID++
		glog.V(4).Infof("Sending observations to Analyzer in chunks, batch [%d] in progress...", batchID)
		obVals, batchToSend := makeBatch(key, iterator, d.batchSize)
		if len(obVals) == 0 {
			break
		}
		sendErr := d.analyzerTransport.send(batchToSend)
		if sendErr == nil {
			if err := d.store.DeleteValues(key, obVals); err != nil {
				stackdriver.LogCountMetricf(dispatchBucketFailed, "Error in deleting dispatched observations from the store for key: %v", key)
			}
		} else {
			stackdriver.LogCountMetricf(dispatchBucketFailed, "Error in transmitting data to Analyzer for key [%v]: %v", key, sendErr)
		}
		time.Sleep(sleepDuration)
	}
	return nil
}
func ( d * Dispatcher ) dispatchBucket ( key * cobalt . ObservationMetadata , sleepDuration time . Duration ) error { if key == nil { panic ( " " ) } if d == nil { panic ( " " ) } iterator , err := d . store . GetObservations ( key ) if err != nil { stackdriver . LogCountMetricf ( dispatchBucketFailed , " " , key , err ) return err } batchID := 0 for { batchID ++ glog . V ( 4 ) . Infof ( " " , batchID ) obVals , batchToSend := makeBatch ( key , iterator , d . batchSize ) if len ( obVals ) == 0 { break } sendErr := d . analyzerTransport . send ( batchToSend ) if sendErr == nil { if err := d . store . DeleteValues ( key , obVals ) ; err != nil { stackdriver . LogCountMetricf ( dispatchBucketFailed , " " , key ) } } else { stackdriver . LogCountMetricf ( dispatchBucketFailed , " " , key , sendErr ) } time . Sleep ( sleepDuration ) } return nil }
dispatchBucket dispatches the ObservationBatch associated with |key| in
chunks of size |batchSize| to Analyzer using grpc transport.
dispatchBucket dispatches the ObservationBatch associated with |key| in chunks of size |batchSize| to Analyzer using grpc transport .
// Retrieve shuffled bucket from store for the given |key| // Send the shuffled bucket to Analyzer in chunks. If the bucket is too // big, send it in multiple chunks of size |batchSize|. // If makeBatch() returned an empty batch then the iteration is done. // After successful send, delete the observations from the local // datastore. // TODO(ukode): Add retry behaviour for 3 or more attempts or use // exponential backoff for errors relating to network issues.