// deleteOldObservations deletes the observations for a given |key| from the
// store if the age of the observation is greater than the configured value
// |disposalAgeInDays|.
deleteOldObservations deletes the observations for a given |key| from the
store if the age of the observation is greater than the configured value
|disposalAgeInDays|.
// deleteOldObservations deletes the observations for a given |key| from the // store if the age of the observation is greater than the configured value // |disposalAgeInDays| .
func (d *Dispatcher) deleteOldObservations(key *cobalt.ObservationMetadata,
	currentDayIndex uint32, disposalAgeInDays uint32) 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(deleteOldObservationsFailed, "GetObservation call failed for key: %v with error: %v", key, err)
		return nil
	}
	const maxDeleteBatchSize = 1000
	for {
		var staleObVals []*shuffler.ObservationVal
		for iterator.Next() {
			obVal, err := iterator.Get()
			if err != nil {
				stackdriver.LogCountMetricf(deleteOldObservationsFailed, "deleteOldObservations: iterator.Get() returned an error: %v", err)
				continue
			}
			if currentDayIndex-obVal.ArrivalDayIndex > disposalAgeInDays {
				staleObVals = append(staleObVals, obVal)
				if len(staleObVals) == maxDeleteBatchSize {
					break
				}
			}
		}
		if len(staleObVals) == 0 {
			break
		} else if err := d.store.DeleteValues(key, staleObVals); err != nil {
			return fmt.Errorf("Error [%v] in deleting old observations for metadata: %v", err, key)
		}
	}
	return nil
}
func ( d * Dispatcher ) deleteOldObservations ( key * cobalt . ObservationMetadata , currentDayIndex uint32 , disposalAgeInDays uint32 ) error { if key == nil { panic ( " " ) } if d == nil { panic ( " " ) } iterator , err := d . store . GetObservations ( key ) if err != nil { stackdriver . LogCountMetricf ( deleteOldObservationsFailed , " " , key , err ) return nil } const maxDeleteBatchSize = 1000 for { var staleObVals [ ] * shuffler . ObservationVal for iterator . Next ( ) { obVal , err := iterator . Get ( ) if err != nil { stackdriver . LogCountMetricf ( deleteOldObservationsFailed , " " , err ) continue } if currentDayIndex - obVal . ArrivalDayIndex > disposalAgeInDays { staleObVals = append ( staleObVals , obVal ) if len ( staleObVals ) == maxDeleteBatchSize { break } } } if len ( staleObVals ) == 0 { break } else if err := d . store . DeleteValues ( key , staleObVals ) ; err != nil { return fmt . Errorf ( " " , err , key ) } } return nil }
deleteOldObservations deletes the observations for a given |key| from the
store if the age of the observation is greater than the configured value
|disposalAgeInDays|.
deleteOldObservations deletes the observations for a given |key| from the store if the age of the observation is greater than the configured value |disposalAgeInDays| .
// We delete stale Observations iteratively in batches of size at most 1000.