signature:
public class HttpMetricsIngestionServer { /** * Starts the Ingest server
 * @ throws InterruptedException */
public void startServer ( ) throws InterruptedException { } }

implementation:
RouteMatcher router = new RouteMatcher ( ) ; router . get ( "/v1.0" , new DefaultHandler ( ) ) ; router . post ( "/v1.0/multitenant/experimental/metrics" , new HttpMultitenantMetricsIngestionHandler ( processor , timeout , ENABLE_PER_TENANT_METRICS ) ) ; router . post ( "/v1.0/:tenantId/experimental/metrics" , new HttpMetricsIngestionHandler ( processor , timeout , ENABLE_PER_TENANT_METRICS ) ) ; router . post ( "/v1.0/:tenantId/experimental/metrics/statsd" , new HttpAggregatedIngestionHandler ( processor , timeout , ENABLE_PER_TENANT_METRICS ) ) ; router . get ( "/v2.0" , new DefaultHandler ( ) ) ; router . post ( "/v2.0/:tenantId/ingest/multi" , new HttpMultitenantMetricsIngestionHandler ( processor , timeout , ENABLE_PER_TENANT_METRICS ) ) ; router . post ( "/v2.0/:tenantId/ingest" , new HttpMetricsIngestionHandler ( processor , timeout , ENABLE_PER_TENANT_METRICS ) ) ; router . post ( "/v2.0/:tenantId/ingest/aggregated" , new HttpAggregatedIngestionHandler ( processor , timeout , ENABLE_PER_TENANT_METRICS ) ) ; router . post ( "/v2.0/:tenantId/ingest/aggregated/multi" , new HttpAggregatedMultiIngestionHandler ( processor , timeout , ENABLE_PER_TENANT_METRICS ) ) ; router . post ( "/v2.0/:tenantId/events" , getHttpEventsIngestionHandler ( ) ) ; final RouteMatcher finalRouter = router ; log . info ( "Starting metrics listener HTTP server on port {}" , httpIngestPort ) ; ServerBootstrap server = new ServerBootstrap ( ) ; server . group ( acceptorGroup , workerGroup ) . channel ( NioServerSocketChannel . class ) . childHandler ( new ChannelInitializer < SocketChannel > ( ) { @ Override public void initChannel ( SocketChannel channel ) throws Exception { setupPipeline ( channel , finalRouter ) ; } } ) ; Channel channel = server . bind ( new InetSocketAddress ( httpIngestHost , httpIngestPort ) ) . sync ( ) . channel ( ) ; allOpenChannels . add ( channel ) ; // register the tracker MBean for JMX / jolokia
log . info ( "Registering tracker service" ) ; Tracker . getInstance ( ) . register ( ) ; log . info ( "Token search improvements enabled: " + EXP_TOKEN_SEARCH_IMPROVEMENTS ) ;


signature:
public class RollupRunnable { /** * determine which DataType to use for serialization . */
public static Rollup . Type getRollupComputer ( RollupType srcType , Granularity srcGran ) { } }

implementation: switch ( srcType ) { case COUNTER : return Rollup . CounterFromCounter ; case TIMER : return Rollup . TimerFromTimer ; case GAUGE : return Rollup . GaugeFromGauge ; case BF_BASIC : return srcGran == Granularity . FULL ? Rollup . BasicFromRaw : Rollup . BasicFromBasic ; case SET : return Rollup . SetFromSet ; default : break ; } throw new IllegalArgumentException ( String . format ( "Cannot compute rollups for %s from %s" , srcType . name ( ) , srcGran . shortName ( ) ) ) ;


signature:
public class IOContainer { /** * Returns an instance of this class based on what configuration says our
 * driver should be .
 * @ return IOContainer */
public static synchronized IOContainer fromConfig ( ) { } }

implementation: if ( FROM_CONFIG_INSTANCE == null ) { String driver = configuration . getStringProperty ( CoreConfig . CASSANDRA_DRIVER ) ; LOG . info ( String . format ( "Using driver %s" , driver ) ) ; boolean isRecordingDelayedMetrics = configuration . getBooleanProperty ( CoreConfig . RECORD_DELAYED_METRICS ) ; LOG . info ( String . format ( "Recording delayed metrics: %s" , isRecordingDelayedMetrics ) ) ; boolean isDtxIngestBatchEnabled = configuration . getBooleanProperty ( CoreConfig . ENABLE_DTX_INGEST_BATCH ) ; LOG . info ( String . format ( "Datastax Ingest batch enabled: %s" , isDtxIngestBatchEnabled ) ) ; FROM_CONFIG_INSTANCE = new IOContainer ( DriverType . getDriverType ( driver ) , isRecordingDelayedMetrics , isDtxIngestBatchEnabled ) ; } return FROM_CONFIG_INSTANCE ;


signature:
public class ConfigTtlProvider { /** * Helper function to build the ttl mapping . Only insert to the mapping if the value is a valid date .
 * @ param ttlMapBuilder
 * @ param config
 * @ param gran
 * @ param rollupType
 * @ param configKey
 * @ return true if the insertion is successful , false otherwise . */
private boolean put ( ImmutableTable . Builder < Granularity , RollupType , TimeValue > ttlMapBuilder , Configuration config , Granularity gran , RollupType rollupType , TtlConfig configKey ) { } }

implementation: int value ; try { value = config . getIntegerProperty ( configKey ) ; if ( value < 0 ) return false ; } catch ( NumberFormatException ex ) { log . trace ( String . format ( "No valid TTL config set for granularity: %s, rollup type: %s" , gran . name ( ) , rollupType . name ( ) ) , ex ) ; return false ; } ttlMapBuilder . put ( gran , rollupType , new TimeValue ( value , TimeUnit . DAYS ) ) ; return true ;


signature:
public class BaseRollup { /** * merge simple numbers with this rollup . */
protected void computeFromSimpleMetrics ( Points < SimpleNumber > input ) throws IOException { } }

implementation: if ( input == null ) { throw new IOException ( "Null input to create rollup from" ) ; } if ( input . isEmpty ( ) ) { return ; } Map < Long , Points . Point < SimpleNumber > > points = input . getPoints ( ) ; for ( Map . Entry < Long , Points . Point < SimpleNumber > > item : points . entrySet ( ) ) { this . count += 1 ; SimpleNumber numericMetric = item . getValue ( ) . getData ( ) ; average . handleFullResMetric ( numericMetric . getValue ( ) ) ; variance . handleFullResMetric ( numericMetric . getValue ( ) ) ; minValue . handleFullResMetric ( numericMetric . getValue ( ) ) ; maxValue . handleFullResMetric ( numericMetric . getValue ( ) ) ; }


signature:
public class BaseRollup { /** * merge rollups into this rollup . */
protected void computeFromRollupsHelper ( Points < ? extends IBaseRollup > input ) throws IOException { } }

implementation:
if ( input == null ) { throw new IOException ( "Null input to create rollup from" ) ; } if ( input . isEmpty ( ) ) { return ; } // See this and get mind blown :
// http : / / stackoverflow . com / questions / 18907262 / bounded - wildcard - related - compiler - error
Map < Long , ? extends Points . Point < ? extends IBaseRollup > > points = input . getPoints ( ) ; for ( Map . Entry < Long , ? extends Points . Point < ? extends IBaseRollup > > item : points . entrySet ( ) ) { IBaseRollup rollup = item . getValue ( ) . getData ( ) ; if ( ! ( rollup instanceof BaseRollup ) ) { throw new IOException ( "Cannot create BaseRollup from type " + rollup . getClass ( ) . getName ( ) ) ; } IBaseRollup baseRollup = ( IBaseRollup ) rollup ; this . count += baseRollup . getCount ( ) ; average . handleRollupMetric ( baseRollup ) ; variance . handleRollupMetric ( baseRollup ) ; minValue . handleRollupMetric ( baseRollup ) ; maxValue . handleRollupMetric ( baseRollup ) ; }


signature:
public class RollupUtils { /** * For a particular rollupType , determine what is the proper class to do
 * read / write with and return that
 * @ param rollupType
 * @ return */
public static AbstractMetricsRW getMetricsRWForRollupType ( RollupType rollupType ) { } }

implementation: if ( rollupType == null || rollupType == RollupType . BF_BASIC ) { return IOContainer . fromConfig ( ) . getBasicMetricsRW ( ) ; } else { return IOContainer . fromConfig ( ) . getPreAggregatedMetricsRW ( ) ; }


signature:
public class DDelayedLocatorIO { /** * package protect so other classes in this package can call it */
BoundStatement getBoundStatementForLocator ( Granularity granularity , int slot , Locator locator ) { } }

implementation: int shard = Util . getShard ( locator . toString ( ) ) ; return putValue . bind ( SlotKey . of ( granularity , slot , shard ) . toString ( ) , locator . toString ( ) , "" ) ;


signature:
public class OutputFormatter { /** * compute the maximum width for each field across a collection of formatters . */
public static int [ ] computeMaximums ( String [ ] headers , OutputFormatter ... outputs ) { } }

implementation: int [ ] max = new int [ headers . length ] ; for ( int i = 0 ; i < headers . length ; i ++ ) max [ i ] = headers [ i ] . length ( ) ; for ( OutputFormatter output : outputs ) { max [ 0 ] = Math . max ( output . host . length ( ) , max [ 0 ] ) ; for ( int i = 1 ; i < headers . length ; i ++ ) max [ i ] = Math . max ( output . results [ i - 1 ] . length ( ) , max [ i ] ) ; } return max ;


signature:
public class OutputFormatter { /** * formats a header row after maximums have been established . */
public static String formatHeader ( int [ ] maximums , String [ ] headers ) { } }

implementation: StringBuilder sb = new StringBuilder ( ) ; for ( int i = 0 ; i < headers . length ; i ++ ) sb = sb . append ( formatIn ( headers [ i ] , maximums [ i ] , false ) ) . append ( GAP ) ; return sb . toString ( ) ;


signature:
public class OutputFormatter { /** * formats results and sets formattedStrings . */
public static String [ ] format ( int [ ] maximums , OutputFormatter ... outputs ) { } }

implementation: String [ ] formattedStrings = new String [ outputs . length ] ; int pos = 0 ; for ( OutputFormatter output : outputs ) { StringBuilder sb = new StringBuilder ( ) ; sb = sb . append ( formatIn ( output . host , maximums [ 0 ] , false ) ) ; for ( int i = 0 ; i < output . results . length ; i ++ ) sb = sb . append ( GAP ) . append ( formatIn ( output . results [ i ] , maximums [ i + 1 ] , true ) ) ; formattedStrings [ pos ++ ] = sb . toString ( ) ; } return formattedStrings ;


signature:
public class ZKShardLockManager { /** * Registers the different ZooKeeper metrics . */
private void registerMetrics ( final ObjectName nameObj , MetricRegistry reg ) { } }

implementation: reg . register ( MetricRegistry . name ( ZKShardLockManager . class , "Lock Disinterested Time Millis" ) , new JmxAttributeGauge ( nameObj , "LockDisinterestedTimeMillis" ) ) ; reg . register ( MetricRegistry . name ( ZKShardLockManager . class , "Min Lock Hold Time Millis" ) , new JmxAttributeGauge ( nameObj , "MinLockHoldTimeMillis" ) ) ; reg . register ( MetricRegistry . name ( ZKShardLockManager . class , "Seconds Since Last Scavenge" ) , new JmxAttributeGauge ( nameObj , "SecondsSinceLastScavenge" ) ) ; reg . register ( MetricRegistry . name ( ZKShardLockManager . class , "Zk Connection Status" ) , new JmxAttributeGauge ( nameObj , "ZkConnectionStatus" ) { @ Override public Object getValue ( ) { Object val = super . getValue ( ) ; if ( val . equals ( "connected" ) ) { return 1 ; } return 0 ; } } ) ; reg . register ( MetricRegistry . name ( ZKShardLockManager . class , "Held Shards" ) , new Gauge < Integer > ( ) { @ Override public Integer getValue ( ) { return getHeldShards ( ) . size ( ) ; } } ) ; reg . register ( MetricRegistry . name ( ZKShardLockManager . class , "Unheld Shards" ) , new Gauge < Integer > ( ) { @ Override public Integer getValue ( ) { return getUnheldShards ( ) . size ( ) ; } } ) ; reg . register ( MetricRegistry . name ( ZKShardLockManager . class , "Error Shards" ) , new Gauge < Integer > ( ) { @ Override public Integer getValue ( ) { return getErrorShards ( ) . size ( ) ; } } ) ;


signature:
public class ZKShardLockManager { /** * Waits until the zookeeper connection is available .
 * @ param waitTimeSeconds */
@ VisibleForTesting boolean waitForZKConnections ( long waitTimeSeconds ) { } }

implementation: for ( int i = 0 ; i < waitTimeSeconds ; i ++ ) { if ( connected ) { return connected ; } log . debug ( "Waiting for connect" ) ; try { Thread . sleep ( 1000 ) ; } catch ( InterruptedException ex ) { } } return connected ;


signature:
public class ZKShardLockManager { /** * Only called from { @ link # init ( TimeValue ) } .
 * Try to achieve each lock . This will give us a good initial state where each lock is either held or unheld .
 * The purpose is to avoid a situation at startup where a node assumes it can schedule anything ( until a lock
 * is held ) . */
@ VisibleForTesting void prefetchLocks ( ) { } }

implementation: if ( ! connected ) { log . warn ( "Cannot connect to Zookeeper; will not perform initial lock acquisition" ) ; for ( Lock lock : locks . values ( ) ) { lock . connectionLost ( ) ; } } else { log . info ( "Pre-fetching zookeeper locks for shards" ) ; boolean isManagingAllShards = locks . size ( ) >= Constants . NUMBER_OF_SHARDS ; int maxLocksToPrefetch = moreThanHalf ( ) ; if ( isManagingAllShards ) { maxLocksToPrefetch = Constants . NUMBER_OF_SHARDS ; } List < Integer > shards = new ArrayList < Integer > ( locks . keySet ( ) ) ; Collections . shuffle ( shards ) ; int locksObtained = 0 ; for ( int shard : shards ) { try { log . debug ( "Initial lock attempt for shard={}" , shard ) ; final Lock lock = locks . get ( shard ) ; lock . acquire ( ) . get ( ) ; if ( lock . isHeld ( ) && ++ locksObtained >= maxLocksToPrefetch ) { break ; } } catch ( InterruptedException ex ) { log . warn ( "Thread exception while acquiring initial locks: " + ex . getMessage ( ) , ex ) ; } catch ( ExecutionException ex ) { log . error ( "Problem acquiring lock " + shard + " " + ex . getCause ( ) . getMessage ( ) , ex . getCause ( ) ) ; } } log . info ( "Finished pre-fetching zookeeper locks" ) ; }


signature:
public class ZKShardLockManager { /** * Unsafe method for testing . */
@ VisibleForTesting void waitForQuiesceUnsafe ( ) { } }

implementation:
while ( lockTaskCount . get ( ) > 0 ) { log . trace ( "Waiting for quiesce" ) ; try { Thread . sleep ( 100 ) ; } catch ( InterruptedException ignore ) { } } // this time out needs to be longer than ZK _ LOCK _ TIMEOUT for valid tests .
try { Thread . sleep ( 2000 ) ; } catch ( InterruptedException ignore ) { }


signature:
public class SlotKey { /** * Given the encoded slot key , returns the java object of it .
 * For the valid slot keys , this function is inverse of { @ link # toString ( ) } .
 * @ return decoded { @ link SlotKey } , < code > null < / code > if it ' s an invalid slotkey .
 * This function is mostly used in the tests . */
public static SlotKey parse ( String string ) { } }

implementation: String [ ] tokens = string . split ( "," ) ; if ( tokens . length != 3 ) { return null ; } Granularity granularity = Granularity . fromString ( tokens [ 0 ] ) ; if ( granularity == null ) { return null ; } try { int slot = Integer . parseInt ( tokens [ 1 ] ) ; int shard = Integer . parseInt ( tokens [ 2 ] ) ; return of ( granularity , slot , shard ) ; } catch ( IllegalArgumentException e ) { return null ; }


signature:
public class SlotKey { /** * This method creates a collection of slot keys that are within the same
 * timespan of the current slot key but of a finer granularity . For example ,
 * a slot key of granularity { @ link Granularity # MIN _ 20 MIN _ 20 } will have
 * four child keys , each of granularity { @ link Granularity # MIN _ 5 MIN _ 5 } .
 * This method is recursive , so that it also includes the current slot
 * key ' s children and their children , and so on , down to the finest
 * granularity . For example , a slot key of granularity
 * { @ link Granularity # MIN _ 1440 MIN _ 1440 } will have 6 immediate children of
 * granularity { @ link Granularity # MIN _ 240 MIN _ 240 } , 6*4 = 24 descendant slot
 * key of granularity { @ link Granularity # MIN _ 60 MIN _ 60 } , 6*4*3 = 72
 * descendant slot key of granularity { @ link Granularity # MIN _ 20 MIN _ 20 } ,
 * 6*4*3*4 = 288 descendant slot key of granularity
 * { @ link Granularity # MIN _ 5 MIN _ 5 } , and 6*4*3*4*1 = 288 descendant slot key of
 * granularity { @ link Granularity # FULL FULL } ; in that case , the top - level
 * call of this method will return a collection 6 + 24 + 72 + 288 + 288 = 678
 * descendants .
 * @ return a Collection of all descendant { @ link SlotKey } s */
public Collection < SlotKey > getChildrenKeys ( ) { } }

implementation: if ( granularity == Granularity . FULL ) { return ImmutableList . of ( ) ; } List < SlotKey > result = new ArrayList < SlotKey > ( ) ; Granularity finer ; try { finer = granularity . finer ( ) ; } catch ( GranularityException e ) { throw new AssertionError ( "Should not occur." ) ; } int factor = finer . numSlots ( ) / granularity . numSlots ( ) ; for ( int i = 0 ; i < factor ; i ++ ) { int childSlot = slot * factor + i ; SlotKey child = SlotKey . of ( finer , childSlot , shard ) ; result . add ( child ) ; result . addAll ( child . getChildrenKeys ( ) ) ; } return result ;


signature:
public class SlotKey { /** * Same as the method { @ link # getChildrenKeys ( ) } except this method returns only
 * the children corresponding to the destination granularity .
 * For example , a slot key of granularity
 * { @ link Granularity # MIN _ 1440 MIN _ 1440 } , for destination granularity of
 * { @ link Granularity # MIN _ 60 MIN _ 60 } , will have 6*4 = 24 children .
 * @ param destGranularity
 * @ return */
public Collection < SlotKey > getChildrenKeys ( Granularity destGranularity ) { } }

implementation: if ( ! getGranularity ( ) . isCoarser ( destGranularity ) ) { throw new IllegalArgumentException ( String . format ( "Current granularity [%s] must be coarser than the destination granularity [%s]" , getGranularity ( ) , destGranularity ) ) ; } List < SlotKey > result = new ArrayList < SlotKey > ( ) ; for ( SlotKey slotKey : this . getChildrenKeys ( ) ) { if ( slotKey . getGranularity ( ) . equals ( destGranularity ) ) { result . add ( slotKey ) ; } } return Collections . unmodifiableList ( result ) ;


signature:
public class SlotKey { /** * This method would extrapolate a given slotkey to a corresponding parent slotkey of a given ( higher ) granularity .
 * For example : For a given slotkey ' metrics _ 5m , 1,30 ' which corresponds to granularity = 5m , slot = 1 , shard = 30 , if we
 * extrapolate this to a destination granularity of 20m , this would return ' metrics _ 20m , 0,30 ' , which
 * corresponds to granularity = 20m , slot = 0 , shard = 30 which is a parent key of the given slotkey
 * @ param destGranularity
 * @ return */
public SlotKey extrapolate ( Granularity destGranularity ) { } }

implementation: if ( destGranularity . equals ( this . getGranularity ( ) ) ) { return this ; } if ( ! destGranularity . isCoarser ( getGranularity ( ) ) ) { throw new IllegalArgumentException ( "Destination granularity must be coarser than the current granularity" ) ; } int factor = getGranularity ( ) . numSlots ( ) / destGranularity . numSlots ( ) ; int parentSlot = getSlot ( ) / factor ; return SlotKey . of ( destGranularity , parentSlot , getShard ( ) ) ;


signature:
public class ThreadPoolBuilder { /** * Set the threadpool name . Used to generate metric names and thread names . */
public ThreadPoolBuilder withName ( String name ) { } }

implementation:
// ensure we ' ve got a spot to put the thread id .
if ( ! name . contains ( "%d" ) ) { name = name + "-%d" ; } nameMap . putIfAbsent ( name , new AtomicInteger ( 0 ) ) ; int id = nameMap . get ( name ) . incrementAndGet ( ) ; this . poolName = String . format ( name , id ) ; if ( id > 1 ) { this . threadNameFormat = name . replace ( "%d" , id + "-%d" ) ; } else { this . threadNameFormat = name ; } return this ;


signature:
public class HttpAggregatedMultiIngestionHandler { /** * our own stuff . */
@ Override public void handle ( ChannelHandlerContext ctx , FullHttpRequest request ) { } }

implementation:
Tracker . getInstance ( ) . track ( request ) ; requestCount . inc ( ) ; final Timer . Context timerContext = handlerTimer . time ( ) ; long ingestTime = clock . now ( ) . getMillis ( ) ; // this is all JSON .
String body = null ; try { String submitterTenantId = request . headers ( ) . get ( HttpMetricsIngestionServer . TENANT_ID_HEADER ) ; body = request . content ( ) . toString ( Constants . DEFAULT_CHARSET ) ; List < AggregatedPayload > bundleList = createBundleList ( body ) ; if ( bundleList . size ( ) > 0 ) { // has aggregated metric bundle in body
// convert and add metric bundle to MetricsCollection if valid
MetricsCollection collection = new MetricsCollection ( ) ; List < ErrorResponse . ErrorData > errors = new ArrayList < ErrorResponse . ErrorData > ( ) ; // for each metric bundle
int delayedMetricsCount = 0 ; int metricsCount = 0 ; for ( AggregatedPayload bundle : bundleList ) { // validate , convert , and add to collection
List < ErrorResponse . ErrorData > bundleValidationErrors = bundle . getValidationErrors ( ) ; if ( bundleValidationErrors . isEmpty ( ) ) { // no validation error , add to collection
collection . add ( PreaggregateConversions . buildMetricsCollection ( bundle ) ) ; } else { // failed validation , add to error
errors . addAll ( bundleValidationErrors ) ; } if ( bundle . hasDelayedMetrics ( ingestTime ) ) { Tracker . getInstance ( ) . trackDelayedAggregatedMetricsTenant ( bundle . getTenantId ( ) , bundle . getTimestamp ( ) , bundle . getDelayTime ( ingestTime ) , bundle . getAllMetricNames ( ) ) ; bundle . markDelayMetricsReceived ( ingestTime ) ; delayedMetricsCount += bundle . getAllMetricNames ( ) . size ( ) ; } else { metricsCount += bundle . getAllMetricNames ( ) . size ( ) ; } } // if has validation errors and no valid metrics
if ( ! errors . isEmpty ( ) && collection . size ( ) == 0 ) { // return BAD _ REQUEST and error
DefaultHandler . sendErrorResponse ( ctx , request , errors , HttpResponseStatus . BAD_REQUEST ) ; return ; } // process valid metrics in collection
ListenableFuture < List < Boolean > > futures = processor . apply ( collection ) ; List < Boolean > persisteds = futures . get ( timeout . getValue ( ) , timeout . getUnit ( ) ) ; for ( Boolean persisted : persisteds ) { if ( ! persisted ) { DefaultHandler . sendErrorResponse ( ctx , request , "Internal error persisting data" , HttpResponseStatus . INTERNAL_SERVER_ERROR ) ; return ; } } recordPerTenantMetrics ( submitterTenantId , metricsCount , delayedMetricsCount ) ; // return OK or MULTI _ STATUS response depending if there were validation errors
if ( errors . isEmpty ( ) ) { // no validation error , response OK
DefaultHandler . sendResponse ( ctx , request , null , HttpResponseStatus . OK ) ; return ; } else { // has some validation errors , response MULTI _ STATUS
DefaultHandler . sendErrorResponse ( ctx , request , errors , HttpResponseStatus . MULTI_STATUS ) ; return ; } } else { // no aggregated metric bundles in body , response OK
DefaultHandler . sendResponse ( ctx , request , "No valid metrics" , HttpResponseStatus . BAD_REQUEST ) ; return ; } } catch ( JsonParseException ex ) { log . debug ( String . format ( "BAD JSON: %s" , body ) ) ; DefaultHandler . sendErrorResponse ( ctx , request , ex . getMessage ( ) , HttpResponseStatus . BAD_REQUEST ) ; } catch ( InvalidDataException ex ) { log . debug ( String . format ( "Invalid request body: %s" , body ) ) ; DefaultHandler . sendErrorResponse ( ctx , request , ex . getMessage ( ) , HttpResponseStatus . BAD_REQUEST ) ; } catch ( TimeoutException ex ) { DefaultHandler . sendErrorResponse ( ctx , request , "Timed out persisting metrics" , HttpResponseStatus . ACCEPTED ) ; } catch ( Exception ex ) { log . debug ( String . format ( "Exception processing: %s" , body ) ) ; log . error ( "Other exception while trying to parse content" , ex ) ; DefaultHandler . sendErrorResponse ( ctx , request , "Internal error saving data" , HttpResponseStatus . INTERNAL_SERVER_ERROR ) ; } finally { timerContext . stop ( ) ; requestCount . dec ( ) ; }


signature:
public class MetricIndexData { /** * Compares actualDocCount and total docCount of its immediate children of an index
 * to determine if the metric index is a complete metric name or not .
 * For the ES response shown in class description , for a baseLevel of 2,
 * foo . bar . baz has actualDocCount of 2 , but total doc count of all its children ,
 * which in this case is only foo . bar . baz , is 1 . So foo . bar . baz must be metric by itself
 * @ param metricIndexMap
 * @ return */
private Set < String > getCompleteMetricNames ( Map < String , MetricIndexDocCount > metricIndexMap ) { } }

implementation:
Set < String > completeMetricNames = new HashSet < String > ( ) ; for ( Map . Entry < String , MetricIndexDocCount > entry : metricIndexMap . entrySet ( ) ) { MetricIndexDocCount metricIndexDocCount = entry . getValue ( ) ; if ( metricIndexDocCount != null ) { // if total doc count is greater than its children docs , its a complete metric name
if ( metricIndexDocCount . actualDocCount > 0 && metricIndexDocCount . actualDocCount > metricIndexDocCount . childrenTotalDocCount ) { completeMetricNames . add ( entry . getKey ( ) ) ; } } } return Collections . unmodifiableSet ( completeMetricNames ) ;


signature:
public class ShardStateManager { /** * Side effect : mark dirty slots as clean */
protected Map < Granularity , Map < Integer , UpdateStamp > > getDirtySlotsToPersist ( int shard ) { } }

implementation:
Map < Granularity , Map < Integer , UpdateStamp > > slotTimes = new HashMap < Granularity , Map < Integer , UpdateStamp > > ( ) ; int numUpdates = 0 ; for ( Granularity gran : Granularity . rollupGranularities ( ) ) { Map < Integer , UpdateStamp > dirty = getSlotStateManager ( shard , gran ) . getDirtySlotStampsAndMarkClean ( ) ; slotTimes . put ( gran , dirty ) ; if ( dirty . size ( ) > 0 ) { numUpdates += dirty . size ( ) ; } } if ( numUpdates > 0 ) { // for updates that come by way of scribe , you ' ll typically see 5 as the number of updates ( one for
// each granularity ) . On rollup slaves the situation is a bit different . You ' ll see only the slot
// of the granularity just written to marked dirty ( so 1 ) .
log . debug ( "Found {} dirty slots for shard {}" , new Object [ ] { numUpdates , shard } ) ; return slotTimes ; } return null ;


signature:
public class Token { /** * This method return list of tokens with their parents for a current Discovery object .
 * For example : A locator of 1111 : a . b . c . d would generate the following tokens
 * Token { token = ' a ' , parent = ' ' , isLeaf = false , id = ' 11111 : a ' , locator = 11111 . a . b . c . d }
 * Token { token = ' b ' , parent = ' a ' , isLeaf = false , id = ' 11111 : a . b ' , locator = 11111 . a . b . c . d }
 * Token { token = ' c ' , parent = ' a . b ' , isLeaf = false , id = ' 11111 : a . b . c ' , locator = 11111 . a . b . c . d }
 * Token { token = ' d ' , parent = ' a . b . c ' , isLeaf = true , id = ' 11111 : a . b . c . d : $ ' , locator = 11111 . a . b . c . d }
 * @ return */
public static List < Token > getTokens ( Locator locator ) { } }

implementation: if ( StringUtils . isEmpty ( locator . getMetricName ( ) ) || StringUtils . isEmpty ( locator . getTenantId ( ) ) ) return new ArrayList < > ( ) ; String [ ] tokens = locator . getMetricName ( ) . split ( Locator . METRIC_TOKEN_SEPARATOR_REGEX ) ; return IntStream . range ( 0 , tokens . length ) . mapToObj ( index -> new Token ( locator , tokens , index ) ) . collect ( toList ( ) ) ;


signature:
public class DAbstractMetricIO { /** * Asynchronously insert a rolled up metric to the appropriate column family
 * for a particular granularity
 * @ param locator
 * @ param collectionTime
 * @ param rollup
 * @ param granularity
 * @ return */
public ResultSetFuture putAsync ( Locator locator , long collectionTime , Rollup rollup , Granularity granularity , int ttl ) { } }

implementation:
Session session = DatastaxIO . getSession ( ) ; // we use batch statement here in case sub classes
// override the addRollupToBatch ( ) and provide
// multiple statements
BatchStatement batch = new BatchStatement ( ) ; addRollupToBatch ( batch , locator , rollup , collectionTime , granularity , ttl ) ; Collection < Statement > statements = batch . getStatements ( ) ; if ( statements . size ( ) == 1 ) { Statement oneStatement = statements . iterator ( ) . next ( ) ; return session . executeAsync ( oneStatement ) ; } else { LOG . debug ( String . format ( "Using BatchStatement for %d statements" , statements . size ( ) ) ) ; return session . executeAsync ( batch ) ; }


signature:
public class DAbstractMetricIO { /** * Fetch rollup objects for a { @ link com . rackspacecloud . blueflood . types . Locator }
 * from the specified column family and range .
 * @ param locator
 * @ param columnFamily
 * @ param range
 * @ return */
protected < T extends Object > Table < Locator , Long , T > getRollupsForLocator ( final Locator locator , String columnFamily , Range range ) { } }

implementation: return getValuesForLocators ( new ArrayList < Locator > ( ) { { add ( locator ) ; } } , columnFamily , range ) ;


signature:
public class DAbstractMetricIO { /** * Fetch values for a list of { @ link com . rackspacecloud . blueflood . types . Locator }
 * from the specified column family and range .
 * This is a base behavior for most rollup types . IO subclasses can override
 * this behavior as they see fit .
 * @ param locators
 * @ param columnFamily
 * @ param range
 * @ return */
protected < T extends Object > Table < Locator , Long , T > getValuesForLocators ( final List < Locator > locators , String columnFamily , Range range ) { } }

implementation: Table < Locator , Long , T > locatorTimestampRollup = HashBasedTable . create ( ) ; Map < Locator , List < ResultSetFuture > > resultSetFuturesMap = selectForLocatorListAndRange ( columnFamily , locators , range ) ; for ( Map . Entry < Locator , List < ResultSetFuture > > entry : resultSetFuturesMap . entrySet ( ) ) { Locator locator = entry . getKey ( ) ; List < ResultSetFuture > futures = entry . getValue ( ) ; Table < Locator , Long , T > result = toLocatorTimestampValue ( futures , locator , columnFamily , range ) ; locatorTimestampRollup . putAll ( result ) ; } return locatorTimestampRollup ;


signature:
public class DAbstractMetricIO { /** * Add a { @ link com . datastax . driver . core . PreparedStatement } statement to the
 * { @ link com . datastax . driver . core . BatchStatement } to insert this Rollup
 * object to metrics _ preaggregated _ { granularity } column family
 * @ param batch
 * @ param locator
 * @ param collectionTime
 * @ param granularity
 * @ param ttl */
protected void addRollupToBatch ( BatchStatement batch , Locator locator , Rollup rollup , long collectionTime , Granularity granularity , int ttl ) { } }

implementation: Statement statement = createStatement ( locator , collectionTime , rollup , granularity , ttl ) ; batch . add ( statement ) ;


signature:
public class DAbstractMetricIO { /** * Asynchronously execute select statements against the specified
 * column family for a specific list of
 * { @ link com . rackspacecloud . blueflood . types . Locator } and
 * { @ link com . rackspacecloud . blueflood . types . Range }
 * @ param locators
 * @ param range
 * @ return a map of Locator - > a list of ResultSetFuture */
protected Map < Locator , List < ResultSetFuture > > selectForLocatorListAndRange ( String columnFamilyName , List < Locator > locators , Range range ) { } }

implementation: Map < Locator , List < ResultSetFuture > > locatorFuturesMap = new HashMap < Locator , List < ResultSetFuture > > ( ) ; for ( Locator locator : locators ) { List < ResultSetFuture > existing = locatorFuturesMap . get ( locator ) ; if ( existing == null ) { existing = new ArrayList < ResultSetFuture > ( ) ; locatorFuturesMap . put ( locator , existing ) ; } existing . addAll ( selectForLocatorAndRange ( columnFamilyName , locator , range ) ) ; } return locatorFuturesMap ;


signature:
public class DAbstractMetricIO { /** * Execute a select statement against the specified column family for a specific
 * { @ link com . rackspacecloud . blueflood . types . Locator } and
 * { @ link com . rackspacecloud . blueflood . types . Range }
 * @ param columnFamily the column family name to do select against
 * @ param locator
 * @ param range
 * @ return */
protected List < ResultSetFuture > selectForLocatorAndRange ( String columnFamily , Locator locator , Range range ) { } }

implementation: List < ResultSetFuture > resultsFutures = new ArrayList < ResultSetFuture > ( ) ; PreparedStatement statement = metricsCFPreparedStatements . cfNameToSelectStatement . get ( columnFamily ) ; resultsFutures . add ( session . executeAsync ( statement . bind ( locator . toString ( ) , range . getStart ( ) , range . getStop ( ) ) ) ) ; return resultsFutures ;


signature:
public class DAbstractMetricIO { /** * Give a { @ link com . datastax . driver . core . ResultSetFuture } , get
 * the corresponding data from it and return it as a
 * Table of locator , long and rollup . */
public < T extends Object > Table < Locator , Long , T > toLocatorTimestampValue ( List < ResultSetFuture > futures , Locator locator , String columnFamily , Range range ) { } }

implementation:
Table < Locator , Long , T > locatorTimestampRollup = HashBasedTable . create ( ) ; for ( ResultSetFuture future : futures ) { try { List < Row > rows = future . getUninterruptibly ( ) . all ( ) ; // we only want to count the number of points we
// get when we ' re querying the metrics _ full
if ( StringUtils . isNotEmpty ( columnFamily ) && columnFamily . equals ( CassandraModel . CF_METRICS_FULL_NAME ) ) { Instrumentation . getRawPointsIn5MinHistogram ( ) . update ( rows . size ( ) ) ; } for ( Row row : rows ) { String key = row . getString ( DMetricsCFPreparedStatements . KEY ) ; Locator loc = Locator . createLocatorFromDbKey ( key ) ; Long hash = row . getLong ( DMetricsCFPreparedStatements . COLUMN1 ) ; locatorTimestampRollup . put ( loc , hash , ( T ) fromByteBuffer ( row . getBytes ( DMetricsCFPreparedStatements . VALUE ) ) ) ; } } catch ( Exception ex ) { Instrumentation . markReadError ( ) ; LOG . error ( String . format ( "error reading metric for locator %s, column family '%s', range %s" , locator , columnFamily , range . toString ( ) ) , ex ) ; } } return locatorTimestampRollup ;


signature:
public class Granularity { /** * We need to derive ranges ( actual times ) from slots ( which are fixed integers that wrap ) when we discover a late
 * slot . These ranges can be derived from a reference point ( which is usually something like now ) .
 * @ param slot
 * @ param referenceMillis
 * @ return */
public Range deriveRange ( int slot , long referenceMillis ) { } }

implementation:
// referenceMillis refers to the current time in reference to the range we want to generate from the supplied
// slot . This implies that the range we wish to return is before slot ( reference ) . allow for slot wrapping .
referenceMillis = snapMillis ( referenceMillis ) ; int refSlot = slot ( referenceMillis ) ; int slotDiff = slot > refSlot ? ( numSlots ( ) - slot + refSlot ) : ( refSlot - slot ) ; long rangeStart = referenceMillis - slotDiff * milliseconds ( ) ; return new Range ( rangeStart , rangeStart + milliseconds ( ) - 1 ) ;


signature:
public class Granularity { /** * Return granularity that maps most closely to requested number of points ,
 * using the algorithm specified in the
 * { @ code GET _ BY _ POINTS _ SELECTION _ ALGORITHM } config value . See
 * { @ link # granularityFromPointsInInterval ( String , long , long , int , String , long ) } .
 * @ param from beginning of interval ( millis )
 * @ param to end of interval ( millis )
 * @ param points count of desired data points
 * @ return */
public static Granularity granularityFromPointsInInterval ( String tenantid , long from , long to , int points ) { } }

implementation: return granularityFromPointsInInterval ( tenantid , from , to , points , GET_BY_POINTS_SELECTION_ALGORITHM , GET_BY_POINTS_ASSUME_INTERVAL , DEFAULT_TTL_COMPARISON_SOURCE ) ;


signature:
public class Granularity { /** * { @ code ttlComparisonClock } defaults to { @ link # DEFAULT _ TTL _ COMPARISON _ SOURCE } .
 * @ see # granularityFromPointsInInterval ( String , long , long , int , String , long , Clock ) */
public static Granularity granularityFromPointsInInterval ( String tenantid , long from , long to , int points , String algorithm , long assumedIntervalMillis ) { } }

implementation: return granularityFromPointsInInterval ( tenantid , from , to , points , algorithm , assumedIntervalMillis , DEFAULT_TTL_COMPARISON_SOURCE ) ;


signature:
public class Granularity { /** * Return granularity that maps most closely to requested number of points based on
 * provided selection algorithm
 * @ param from beginning of interval ( millis )
 * @ param to end of interval ( millis )
 * @ param points count of desired data points
 * @ param algorithm the algorithm to use . Valid values are
 * { @ code " GEOMETRIC " } , { @ code " LINEAR " } , and
 * { @ code " LESSTHANEQUAL " } . Any other value is treated as
 * { @ code " GEOMETRIC " } .
 * @ param assumedIntervalMillis FULL resolution is tricky because we don ' t
 * know the period of check in question .
 * Assume the minimum period and go from there .
 * NOTE : This value and description was from a
 * previous version of the software , and the
 * original author is no longer available to
 * clarify . The exact purpose of this
 * parameter is not yet fully understood .
 * @ param ttlComparisonClock The GEOMETRIC algorithm checks the specified
 * interval against a TTL . Normally , the TTL is
 * measured relative to the current point in
 * time . For testing purposes , an alternate time
 * source can be provided , to measure the TTL
 * relative to a different point in time .
 * @ return */
public static Granularity granularityFromPointsInInterval ( String tenantid , long from , long to , int points , String algorithm , long assumedIntervalMillis , Clock ttlComparisonClock ) { } }

implementation: if ( from >= to ) { throw new RuntimeException ( "Invalid interval specified for fromPointsInInterval" ) ; } double requestedDuration = to - from ; if ( algorithm . startsWith ( "GEOMETRIC" ) ) return granularityFromPointsGeometric ( tenantid , from , to , requestedDuration , points , assumedIntervalMillis , ttlComparisonClock ) ; else if ( algorithm . startsWith ( "LINEAR" ) ) return granularityFromPointsLinear ( requestedDuration , points , assumedIntervalMillis ) ; else if ( algorithm . startsWith ( "LESSTHANEQUAL" ) ) return granularityFromPointsLessThanEqual ( requestedDuration , points , assumedIntervalMillis ) ; return granularityFromPointsGeometric ( tenantid , from , to , requestedDuration , points , assumedIntervalMillis , ttlComparisonClock ) ;


signature:
public class Granularity { /** * Find the granularity in the interval that will yield a number of data points that are
 * closest to the requested points but < = requested points .
 * @ param requestedDuration
 * @ param points
 * @ return */
private static Granularity granularityFromPointsLessThanEqual ( double requestedDuration , int points , long assumedIntervalMillis ) { } }

implementation:
Granularity gran = granularityFromPointsLinear ( requestedDuration , points , assumedIntervalMillis ) ; if ( requestedDuration / gran . milliseconds ( ) > points ) { try { gran = gran . coarser ( ) ; } catch ( GranularityException e ) { /* do nothing , already at 1440m */
} } return gran ;


signature:
public class Granularity { /** * Find the granularity in the interval that will yield a number of data points that are close to $ points
 * in terms of linear distance .
 * @ param requestedDuration
 * @ param points
 * @ return */
private static Granularity granularityFromPointsLinear ( double requestedDuration , int points , long assumedIntervalMillis ) { } }

implementation: int closest = Integer . MAX_VALUE ; int diff = 0 ; Granularity gran = null ; for ( Granularity g : Granularity . granularities ( ) ) { if ( g == Granularity . FULL ) diff = ( int ) Math . abs ( points - ( requestedDuration / assumedIntervalMillis ) ) ; else diff = ( int ) Math . abs ( points - ( requestedDuration / g . milliseconds ( ) ) ) ; if ( diff < closest ) { closest = diff ; gran = g ; } else { break ; } } return gran ;


signature:
public class RollupService { /** * set the server time in millis . */
public synchronized void setServerTime ( long millis ) { } }

implementation: log . info ( "Manually setting server time to {}  {}" , millis , new java . util . Date ( millis ) ) ; context . setCurrentTimeMillis ( millis ) ;


signature:
public class RollupService { /** * Add a shard to be managed ( via JMX )
 * @ param shard shard to be added */
public void addShard ( Integer shard ) { } }

implementation: if ( ! shardStateManager . getManagedShards ( ) . contains ( shard ) ) context . addShard ( shard ) ;


signature:
public class RollupService { /** * Remove a shard from being managed ( via JMX )
 * @ param shard shard to be removed */
public void removeShard ( Integer shard ) { } }

implementation: if ( shardStateManager . getManagedShards ( ) . contains ( shard ) ) context . removeShard ( shard ) ;


signature:
public class TokenDiscoveryWriter { /** * For all of these { @ link Locator } ' s , get a unique list of { @ link Token } ' s which are not in token cache
 * @ param locators
 * @ return */
static Set < Token > getUniqueTokens ( final List < Locator > locators ) { } }

implementation: return Token . getUniqueTokens ( locators . stream ( ) ) . filter ( token -> ! TokenCache . getInstance ( ) . isTokenCurrent ( token ) ) . collect ( toSet ( ) ) ;


signature:
public class TokenDiscoveryWriter { /** * Get all { @ link Locator } ' s corresponding to the metrics which are not current .
 * @ param input
 * @ return */
static List < Locator > getLocators ( final List < List < IMetric > > input ) { } }

implementation:
// converting list of lists of metrics to flat list of locators that are not current .
return input . stream ( ) . flatMap ( List :: stream ) . map ( IMetric :: getLocator ) . filter ( locator -> ! LocatorCache . getInstance ( ) . isLocatorCurrentInTokenDiscoveryLayer ( locator ) ) . collect ( toList ( ) ) ;


signature:
public class TokenDiscoveryWriter { /** * For a given batch of metrics , insert unique tokens using { @ link TokenDiscoveryIO } .
 * This methods has the following steps .
 * 1 ) For a batch of metrics , get all locators , which are not current by calling { @ link # getLocators ( List ) } .
 * 2 ) For all these locators , get unique tokens , which are not current , by calling { @ link # getUniqueTokens ( List ) }
 * 3 ) insert tokens
 * 4 ) After successful insertion , update both { @ link LocatorCache } and { @ link TokenCache }
 * @ param input
 * @ return */
public ListenableFuture < Boolean > processTokens ( final List < List < IMetric > > input ) { } }

implementation:
return getThreadPool ( ) . submit ( ( ) -> { boolean success = true ; List < Locator > locators = getLocators ( input ) ; List < Token > tokens = new ArrayList < > ( ) ; tokens . addAll ( getUniqueTokens ( locators ) ) ; if ( tokens . size ( ) > 0 ) { for ( TokenDiscoveryIO io : tokenDiscoveryIOs ) { try { io . insertDiscovery ( tokens ) ; } catch ( Exception ex ) { getLogger ( ) . error ( ex . getMessage ( ) , ex ) ; writeErrorMeters . get ( io . getClass ( ) ) . mark ( ) ; success = false ; } } } if ( success && tokens . size ( ) > 0 ) { tokens . stream ( ) . filter ( token -> ! token . isLeaf ( ) ) // do not cache leaf nodes
. forEach ( token -> { // updating token cache
TokenCache . getInstance ( ) . setTokenCurrent ( token ) ; } ) ; locators . stream ( ) . forEach ( locator -> { // updating locator cache
LocatorCache . getInstance ( ) . setLocatorCurrentInTokenDiscoveryLayer ( locator ) ; } ) ; } return success ; } ) ;


signature:
public class InstrumentedThreadPoolExecutor { /** * Given a { @ link ThreadPoolExecutor } , attach various { @ link Gauge } s against its monitoring
 * properties .
 * @ param executor { @ link ThreadPoolExecutor } to monitor .
 * @ param threadPoolName a unique name for this thread pool . */
public static void instrument ( final ThreadPoolExecutor executor , String threadPoolName ) { } }

implementation: MetricRegistry registry = Metrics . getRegistry ( ) ; registry . register ( name ( threadPoolName , "queue-size" ) , new Gauge < Integer > ( ) { @ Override public Integer getValue ( ) { return executor . getQueue ( ) . size ( ) ; } } ) ; registry . register ( name ( threadPoolName , "queue-max" ) , new Gauge < Integer > ( ) { @ Override public Integer getValue ( ) { return executor . getQueue ( ) . size ( ) + executor . getQueue ( ) . remainingCapacity ( ) ; } } ) ; registry . register ( name ( threadPoolName , "threadpool-active" ) , new Gauge < Integer > ( ) { @ Override public Integer getValue ( ) { return executor . getActiveCount ( ) ; } } ) ; registry . register ( name ( threadPoolName , "threadpool-max" ) , new Gauge < Integer > ( ) { @ Override public Integer getValue ( ) { return executor . getMaximumPoolSize ( ) ; } } ) ;


signature:
public class IOConfig { /** * Retrieves the set of unique Cassandra hosts from configuration file .
 * If a single host appears multiple times in the configuration , only one will
 * be listed .
 * @ return set of unique Cassandra hosts */
public Set < String > getUniqueHosts ( ) { } }

implementation: Set < String > uniqueHosts = new HashSet < String > ( ) ; Collections . addAll ( uniqueHosts , config . getStringProperty ( CoreConfig . CASSANDRA_HOSTS ) . split ( "," ) ) ; return uniqueHosts ;


signature:
public class IOConfig { /** * Retrieves a set of unique Cassandra hosts with Binary Transport protocol
 * enabled from configuration file . This may or may not be different than
 * { @ link # getUniqueHosts ( ) }
 * @ return set of unique Cassandra hosts with binary transport protocol enabled */
protected Set < String > getUniqueBinaryTransportHosts ( ) { } }

implementation: Set < String > uniqueHosts = new HashSet < String > ( ) ; Collections . addAll ( uniqueHosts , config . getStringProperty ( CoreConfig . CASSANDRA_BINXPORT_HOSTS ) . split ( "," ) ) ; return uniqueHosts ;


signature:
public class IOConfig { /** * Retrieves a set of unique Cassandra hosts with Binary Transport protocol
 * enabled from configuration file as a set of { @ link java . net . InetSocketAddress }
 * objects .
 * @ return set of InetSocketAddress objects containing unique Cassandra hosts
 * with binary transport protocol enabled */
public Set < InetSocketAddress > getUniqueBinaryTransportHostsAsInetSocketAddresses ( ) { } }

implementation: Set < String > hosts = getUniqueBinaryTransportHosts ( ) ; Set < InetSocketAddress > inetAddresses = new HashSet < InetSocketAddress > ( ) ; for ( String host : hosts ) { String [ ] parts = host . split ( ":" ) ; InetSocketAddress inetSocketAddress ; if ( parts . length == 1 ) { inetSocketAddress = new InetSocketAddress ( parts [ 0 ] , config . getIntegerProperty ( CoreConfig . CASSANDRA_BINXPORT_PORT ) ) ; } else { inetSocketAddress = new InetSocketAddress ( parts [ 0 ] , Integer . parseInt ( parts [ 1 ] ) ) ; } inetAddresses . add ( inetSocketAddress ) ; } return inetAddresses ;


signature:
public class IOConfig { /** * Calculates the number of max connections per Cassandra hosts
 * @ param numHosts the number of Cassandra hosts in the cluster
 * @ return the number of max connections */
public int getMaxConnPerHost ( int numHosts ) { } }

implementation: int maxConns = config . getIntegerProperty ( CoreConfig . MAX_CASSANDRA_CONNECTIONS ) ; return maxConns / numHosts + ( maxConns % numHosts == 0 ? 0 : 1 ) ;


signature:
public class BaseRollupSerDes { /** * Serialize without sum attribute . This is used by { @ link GaugeSerDes }
 * @ param baseRollup
 * @ param protobufOut
 * @ throws IOException */
protected void serializeRollupV1 ( BaseRollup baseRollup , CodedOutputStream protobufOut ) throws IOException { } }

implementation: protobufOut . writeRawByte ( Constants . VERSION_1_ROLLUP ) ; serializeBaseRollupHelper ( baseRollup , protobufOut ) ;


signature:
public class ElasticsearchRestHelper { /** * For a given glob , gives regex for { @ code Locator . METRIC _ TOKEN _ SEPARATOR } separated tokens
 * For example :
 * globPattern of foo . * . * would produce a regex foo \ . [ ^ . ] + \ . [ ^ . ] +
 * globPattern of foo . b * . * would produce a regex foo \ . b [ ^ . ] * \ . [ ^ . ] +
 * @ param globPattern
 * @ return */
protected String getRegexToHandleTokens ( GlobPattern globPattern ) { } }

implementation: String [ ] queryRegexParts = globPattern . compiled ( ) . toString ( ) . split ( "\\\\." ) ; return Arrays . stream ( queryRegexParts ) . map ( this :: convertRegexToCaptureUptoNextToken ) . collect ( joining ( Locator . METRIC_TOKEN_SEPARATOR_REGEX ) ) ;


signature:
public class ElasticsearchRestHelper { /** * If index ( ) fails for whatever reason , it always throws IOException because indexing failed for Elasticsearch . */
public void index ( final String urlFormat , final String bulkString ) throws IOException { } }

implementation:
String tempUrl = String . format ( urlFormat , getNextBaseUrl ( ) ) ; HttpEntity entity = new NStringEntity ( bulkString , ContentType . APPLICATION_JSON ) ; int statusCode = 0 ; /* Here I am using Queue to keep a round - robin selection of next base url . If current base URL fails for
 whatever reason ( with response = = null ) , then in catch block , I am enqueueing the next base URL so that
 in next iteration call picks up the new URL . If URL works , queue will be empty and loop will break out . */
Queue < String > callQ = new LinkedList < > ( ) ; callQ . add ( tempUrl ) ; int callCount = 0 ; while ( ! callQ . isEmpty ( ) && callCount < MAX_CALL_COUNT ) { callCount ++ ; String url = callQ . remove ( ) ; logger . debug ( "Using url [{}]" , url ) ; HttpPost httpPost = new HttpPost ( url ) ; httpPost . setHeaders ( getHeaders ( ) ) ; httpPost . setEntity ( entity ) ; CloseableHttpResponse response = null ; try { logger . debug ( "ElasticsearchRestHelper.index Thread name in use: [{}]" , Thread . currentThread ( ) . getName ( ) ) ; response = closeableHttpClient . execute ( httpPost ) ; statusCode = response . getStatusLine ( ) . getStatusCode ( ) ; String str = EntityUtils . toString ( response . getEntity ( ) ) ; EntityUtils . consume ( response . getEntity ( ) ) ; if ( statusCode != HttpStatus . SC_OK && statusCode != HttpStatus . SC_CREATED ) { logger . error ( "index method failed with status code: {} and error: {}" , statusCode , str ) ; } } catch ( Exception e ) { if ( response == null ) { logger . error ( "index method failed with message: {}" , e . getMessage ( ) ) ; url = String . format ( urlFormat , getNextBaseUrl ( ) ) ; callQ . add ( url ) ; } else { logger . error ( "index method failed with status code: {} and exception message: {}" , statusCode , e . getMessage ( ) ) ; } } finally { if ( response != null ) { response . close ( ) ; } } } if ( statusCode != HttpStatus . SC_OK && statusCode != HttpStatus . SC_CREATED ) throw new IOException ( "Elasticsearch indexing failed with status code: [" + statusCode + "]" ) ;


signature:
public class CassandraModel { /** * iterate over all column families that store metrics . */
public static Iterable < MetricColumnFamily > getMetricColumnFamilies ( ) { } }

implementation: return new Iterable < MetricColumnFamily > ( ) { @ Override public Iterator < MetricColumnFamily > iterator ( ) { return new Iterator < MetricColumnFamily > ( ) { private int pos = 0 ; @ Override public boolean hasNext ( ) { return pos < METRIC_COLUMN_FAMILES . length ; } @ Override public MetricColumnFamily next ( ) { return METRIC_COLUMN_FAMILES [ pos ++ ] ; } @ Override public void remove ( ) { throw new NoSuchMethodError ( "Not implemented" ) ; } } ; } } ;


signature:
public class Migration { /** * should be populated . */
private static Map < String , Object > parseOptions ( String [ ] args ) { } }

implementation:
final GnuParser parser = new GnuParser ( ) ; final Map < String , Object > options = new HashMap < String , Object > ( ) ; try { final long now = System . currentTimeMillis ( ) ; CommandLine line = parser . parse ( cliOptions , args ) ; options . put ( SRC , line . getOptionValue ( SRC ) ) ; options . put ( DST , line . getOptionValue ( DST ) ) ; // default range is one year ago until now .
options . put ( FROM , line . hasOption ( FROM ) ? parseDateTime ( line . getOptionValue ( FROM ) ) : now - ( 365L * 24L * 60L * 60L * 1000L ) ) ; options . put ( TO , line . hasOption ( TO ) ? parseDateTime ( line . getOptionValue ( TO ) ) : now ) ; options . put ( LIMIT , line . hasOption ( LIMIT ) ? Integer . parseInt ( line . getOptionValue ( LIMIT ) ) : Integer . MAX_VALUE ) ; options . put ( SKIP , line . hasOption ( SKIP ) ? Integer . parseInt ( line . getOptionValue ( SKIP ) ) : 0 ) ; options . put ( BATCH_SIZE , line . hasOption ( BATCH_SIZE ) ? Integer . parseInt ( line . getOptionValue ( BATCH_SIZE ) ) : 100 ) ; // create a mapping of all cf names - > cf .
// then determine which column family to process .
Map < String , ColumnFamily < Locator , Long > > nameToCf = new HashMap < String , ColumnFamily < Locator , Long > > ( ) { { for ( CassandraModel . MetricColumnFamily cf : CassandraModel . getMetricColumnFamilies ( ) ) { put ( cf . getName ( ) , cf ) ; } } } ; if ( nameToCf . get ( line . getOptionValue ( COLUMN_FAMILY ) ) == null ) { throw new ParseException ( "Invalid column family" ) ; } CassandraModel . MetricColumnFamily columnFamily = ( CassandraModel . MetricColumnFamily ) nameToCf . get ( line . getOptionValue ( COLUMN_FAMILY ) ) ; options . put ( COLUMN_FAMILY , columnFamily ) ; options . put ( TTL , line . hasOption ( TTL ) ? Integer . parseInt ( line . getOptionValue ( TTL ) ) : ( int ) ( 5 * columnFamily . getDefaultTTL ( ) . toSeconds ( ) ) ) ; options . put ( READ_THREADS , line . hasOption ( READ_THREADS ) ? Integer . parseInt ( line . getOptionValue ( READ_THREADS ) ) : 1 ) ; options . put ( WRITE_THREADS , line . hasOption ( WRITE_THREADS ) ? Integer . parseInt ( line . getOptionValue ( WRITE_THREADS ) ) : 1 ) ; options . put ( VERIFY , line . hasOption ( VERIFY ) ) ; options . put ( DISCOVER , line . hasOption ( DISCOVER ) ? NodeDiscoveryType . RING_DESCRIBE : NodeDiscoveryType . NONE ) ; options . put ( RATE , line . hasOption ( RATE ) ? Integer . parseInt ( line . getOptionValue ( RATE ) ) : 500 ) ; } catch ( ParseException ex ) { HelpFormatter helpFormatter = new HelpFormatter ( ) ; helpFormatter . printHelp ( "bf-migrate" , cliOptions ) ; System . exit ( - 1 ) ; } return options ;


signature:
public class DTimerIO { /** * Provides a way for the sub class to get a { @ link java . nio . ByteBuffer }
 * representation of a certain Rollup object .
 * @ param value
 * @ return */
@ Override protected ByteBuffer toByteBuffer ( Object value ) { } }

implementation: if ( ! ( value instanceof BluefloodTimerRollup ) ) { throw new IllegalArgumentException ( "toByteBuffer(): expecting BluefloodTimerRollup class but got " + value . getClass ( ) . getSimpleName ( ) ) ; } BluefloodTimerRollup TimerRollup = ( BluefloodTimerRollup ) value ; return serDes . serialize ( TimerRollup ) ;


signature:
public class APreaggregatedMetricsRW { /** * Inserts a collection of metrics to the metrics _ preaggregated _ full column family
 * @ param metrics
 * @ throws IOException */
@ Override public void insertMetrics ( Collection < IMetric > metrics ) throws IOException { } }

implementation: insertMetrics ( metrics , Granularity . FULL ) ;


signature:
public class APreaggregatedMetricsRW { /** * Inserts a collection of metrics to the correct column family based on
 * the specified granularity
 * @ param metrics
 * @ param granularity
 * @ throws IOException */
@ Override public void insertMetrics ( Collection < IMetric > metrics , Granularity granularity ) throws IOException { } }

implementation: try { AstyanaxWriter . getInstance ( ) . insertMetrics ( metrics , CassandraModel . getPreaggregatedColumnFamily ( granularity ) , isRecordingDelayedMetrics , clock ) ; } catch ( ConnectionException ex ) { throw new IOException ( ex ) ; }


signature:
public class APreaggregatedMetricsRW { /** * Fetches { @ link com . rackspacecloud . blueflood . outputs . formats . MetricData } objects for the
 * specified { @ link com . rackspacecloud . blueflood . types . Locator } and
 * { @ link com . rackspacecloud . blueflood . types . Range } from the specified column family
 * @ param locator
 * @ param range
 * @ param gran
 * @ return */
@ Override public MetricData getDatapointsForRange ( final Locator locator , Range range , Granularity gran ) { } }

implementation: return AstyanaxReader . getInstance ( ) . getDatapointsForRange ( locator , range , gran ) ;


signature:
public class APreaggregatedMetricsRW { /** * Fetches { @ link com . rackspacecloud . blueflood . outputs . formats . MetricData } objects for the
 * specified list of { @ link com . rackspacecloud . blueflood . types . Locator } and
 * { @ link com . rackspacecloud . blueflood . types . Range } from the specified column family
 * @ param locators
 * @ param range
 * @ param gran
 * @ return */
@ Override public Map < Locator , MetricData > getDatapointsForRange ( List < Locator > locators , Range range , Granularity gran ) { } }

implementation: return AstyanaxReader . getInstance ( ) . getDatapointsForRange ( locators , range , gran ) ;


signature:
public class APreaggregatedMetricsRW { /** * Fetches a { @ link com . rackspacecloud . blueflood . types . Points } object for a
 * particular locator and rollupType from the specified column family and
 * range
 * @ param locator
 * @ param rollupType
 * @ param range
 * @ param columnFamilyName
 * @ param < T > the type of Rollup object
 * @ return */
@ Override public < T extends Rollup > Points < T > getDataToRollup ( final Locator locator , RollupType rollupType , Range range , String columnFamilyName ) throws IOException { } }

implementation:
ColumnFamily cf = CassandraModel . getColumnFamily ( columnFamilyName ) ; // a quick n dirty hack , this code will go away someday
Granularity granularity = CassandraModel . getGranularity ( cf ) ; Class < ? extends Rollup > rollupClass = RollupType . classOf ( rollupType , granularity ) ; return AstyanaxReader . getInstance ( ) . getDataToRoll ( rollupClass , locator , range , cf ) ;


signature:
public class DSimpleNumberIO { /** * Inserts a raw metric ( not its rolled up value ) to the proper
 * column family
 * @ param metric
 * @ return */
public ResultSetFuture insertRawAsync ( IMetric metric ) { } }

implementation: BoundStatement bound = getBoundStatementForMetric ( metric ) ; return session . executeAsync ( bound ) ;


signature:
public class DSimpleNumberIO { /** * Constructs a { @ link BoundStatement } for a particular metric .
 * The statement could be part of a BatchStatement or it could
 * be executed as is .
 * @ param metric
 * @ return */
public BoundStatement getBoundStatementForMetric ( IMetric metric ) { } }

implementation: return metricsCFPreparedStatements . insertToMetricsBasicFullStatement . bind ( metric . getLocator ( ) . toString ( ) , metric . getCollectionTime ( ) , serDes . serialize ( metric . getMetricValue ( ) ) , metric . getTtlInSeconds ( ) ) ;


signature:
public class LocatorFetchRunnable { /** * This methods gets locators to rollup a slot .
 * During re - rolls , If the granularity of the slot is finer ( or equal ) to the configured
 * DELAYED _ METRICS _ REROLL _ GRANULARITY , we re - roll only the delayed locator ' s corresponding
 * to that slot . if not , we re - roll the locators of the entire shard .
 * @ param executionContext
 * @ param isReroll
 * @ param delayedMetricsRerollGranularity
 * @ param delayedMetricsStorageGranularity
 * @ return */
protected Set < Locator > getLocators ( RollupExecutionContext executionContext , boolean isReroll , Granularity delayedMetricsRerollGranularity , Granularity delayedMetricsStorageGranularity ) { } }

implementation:
Set < Locator > locators = new HashSet < Locator > ( ) ; // if delayed metric tracking is enabled , if its re - roll , if slot granularity is no coarser than
// DELAYED _ METRICS _ REROLL _ GRANULARITY , get delayed locators
if ( RECORD_DELAYED_METRICS && isReroll && ! getGranularity ( ) . isCoarser ( delayedMetricsRerollGranularity ) ) { if ( getGranularity ( ) . isCoarser ( delayedMetricsStorageGranularity ) ) { // For example , if we are re - rolling a 60m slot , and we store delayed metrics at 20m , we need to
// grab delayed metrics for 3 * 20m slots corresponding to the 60m slot .
for ( SlotKey slotKey : parentSlotKey . getChildrenKeys ( delayedMetricsStorageGranularity ) ) { locators . addAll ( getDelayedLocators ( executionContext , slotKey ) ) ; } } else { locators = getDelayedLocators ( executionContext , parentSlotKey . extrapolate ( delayedMetricsStorageGranularity ) ) ; } } else { locators = getLocators ( executionContext ) ; } return locators ;


signature:
public class AbstractElasticIO { /** * This method returns a list of { @ link MetricName } ' s matching the given glob query .
 * for metrics : foo . bar . xxx ,
 * foo . bar . baz . qux ,
 * for query = foo . bar . * , returns the below list of metric names
 * new MetricName ( " foo . bar . xxx " , true ) < - From metric foo . bar . xxx
 * new MetricName ( " foo . bar . baz " , false ) < - From metric foo . bar . baz . qux
 * @ param tenant
 * @ param query is glob representation of hierarchical levels of token . Ex : foo . bar . *
 * @ return
 * @ throws Exception */
public List < MetricName > getMetricNames ( final String tenant , final String query ) throws Exception { } }

implementation:
Timer . Context esMetricNamesQueryTimerCtx = esMetricNamesQueryTimer . time ( ) ; String response ; try { String regexString = regexToGrabCurrentAndNextLevel ( query ) ; // replace one ' \ ' char with two ' \ \ '
regexString = regexString . replaceAll ( "\\\\" , "\\\\\\\\" ) ; response = getMetricNamesFromES ( tenant , regexString ) ; } finally { esMetricNamesQueryTimerCtx . stop ( ) ; } // For example , if query = foo . bar . * , base level is 3 which is equal to the number of tokens in the query .
int baseLevel = getTotalTokens ( query ) ; MetricIndexData metricIndexData = getMetricIndexData ( response , baseLevel ) ; List < MetricName > metricNames = new ArrayList < > ( ) ; // Metric Names matching query which have next level
metricNames . addAll ( metricIndexData . getMetricNamesWithNextLevel ( ) . stream ( ) . map ( x -> new MetricName ( x , false ) ) . collect ( toSet ( ) ) ) ; // complete metric names matching query
metricNames . addAll ( metricIndexData . getCompleteMetricNamesAtBaseLevel ( ) . stream ( ) . map ( x -> new MetricName ( x , true ) ) . collect ( toSet ( ) ) ) ; return metricNames ;


signature:
public class AbstractElasticIO { /** * Performs terms aggregation by metric _ name which returns doc _ count by
 * metric _ name index that matches the given regex .
 * Sample request body :
 * " size " : 0,
 * " query " : {
 * " bool " : {
 * " must " : [ {
 * " term " : {
 * " tenantId " : " ratanasv "
 * " regexp " : {
 * " metric _ name " : {
 * " value " : " < regex > "
 * " aggs " : {
 * " metric _ name _ tokens " : {
 * " terms " : {
 * " field " : " metric _ name " ,
 * " include " : " < regex > " ,
 * " execution _ hint " : " map " ,
 * " size " : 0
 * The two regex expressions used in the query above would be same , one to filter
 * at query level and another to filter the aggregation buckets .
 * Execution hint of " map " works by using field values directly instead of ordinals
 * in order to aggregate data per - bucket
 * @ param tenant
 * @ param regexMetricName
 * @ return */
private String getMetricNamesFromES ( final String tenant , final String regexMetricName ) throws IOException { } }

implementation: String metricNamesFromElasticsearchQueryString = String . format ( queryToFetchMetricNamesFromElasticsearchFormat , tenant , regexMetricName , regexMetricName ) ; return elasticsearchRestHelper . fetchDocuments ( ELASTICSEARCH_INDEX_NAME_READ , ELASTICSEARCH_DOCUMENT_TYPE , tenant , metricNamesFromElasticsearchQueryString ) ;


signature:
public class AbstractElasticIO { /** * Returns regex which could grab metric names from current level to the next level
 * for a given query .
 * ( Some exceptions when query has only one level due to the nature of underlying data )
 * for metrics : foo . bar . baz ,
 * foo . bar . baz . qux ,
 * for query = foo . bar . * , the regex which this method returns will capture the following metric token paths .
 * " foo . bar . baz " < - current level
 * " foo . bar . baz . qux " < - next level
 * @ param query
 * @ return */
protected String regexToGrabCurrentAndNextLevel ( final String query ) { } }

implementation:
if ( StringUtils . isEmpty ( query ) ) { throw new IllegalArgumentException ( "Query(glob) string cannot be null/empty" ) ; } String queryRegex = getRegex ( query ) ; int totalQueryTokens = getTotalTokens ( query ) ; if ( totalQueryTokens == 1 ) { // get metric names which matches the given query and have a next level ,
// Ex : For metric foo . bar . baz . qux , if query = * , we should get foo . bar . We are not
// grabbing 0 level as it will give back bar , baz , qux because of the way data is structured .
String baseRegex = convertRegexToCaptureUptoNextToken ( queryRegex ) ; return baseRegex + METRIC_TOKEN_SEPARATOR_REGEX + REGEX_TO_GRAB_SINGLE_TOKEN ; } else { String [ ] queryRegexParts = queryRegex . split ( "\\\\." ) ; String queryRegexUptoPrevLevel = StringUtils . join ( queryRegexParts , METRIC_TOKEN_SEPARATOR_REGEX , 0 , totalQueryTokens - 1 ) ; String baseRegex = convertRegexToCaptureUptoNextToken ( queryRegexUptoPrevLevel ) ; String queryRegexLastLevel = queryRegexParts [ totalQueryTokens - 1 ] ; String lastTokenRegex = convertRegexToCaptureUptoNextToken ( queryRegexLastLevel ) ; // Ex : For metric foo . bar . baz . qux . xxx , if query = foo . bar . b * , get foo . bar . baz , foo . bar . baz . qux
// In this case baseRegex = " foo . bar " , lastTokenRegex = " b [ ^ . ] * " ' and the final
// regex is foo \ . bar \ . b [ ^ . ] * ( \ . [ ^ . ] * ) { 0,1}
return baseRegex + METRIC_TOKEN_SEPARATOR_REGEX + lastTokenRegex + "(" + METRIC_TOKEN_SEPARATOR_REGEX + REGEX_TO_GRAB_SINGLE_TOKEN + ")" + "{0,1}" ; }


signature:
public class StorageManager { /** * Start background storage management and uploading tasks . */
public synchronized void start ( ) { } }

implementation: if ( uploaderThread != null ) { throw new RuntimeException ( "StorageManager is already started" ) ; } fileUploader = new DoneFileUploader ( ) ; uploaderThread = new Thread ( fileUploader , "StorageManager uploader" ) ; uploaderThread . start ( ) ;


signature:
public class StorageManager { /** * Stop background storage management .
 * @ throws IOException */
public synchronized void stop ( ) throws IOException { } }

implementation: if ( uploaderThread == null ) { throw new RuntimeException ( "Not running" ) ; } uploaderThread . interrupt ( ) ; uploaderThread = null ; fileUploader . shutdown ( ) ;


signature:
public class BasicRollup { /** * merge simple numbers with this rollup . */
protected void computeFromSimpleMetrics ( Points < SimpleNumber > input ) throws IOException { } }

implementation: super . computeFromSimpleMetrics ( input ) ; if ( input . isEmpty ( ) ) { return ; } Map < Long , Points . Point < SimpleNumber > > points = input . getPoints ( ) ; for ( Map . Entry < Long , Points . Point < SimpleNumber > > item : points . entrySet ( ) ) { SimpleNumber numericMetric = item . getValue ( ) . getData ( ) ; sum += numericMetric . getValue ( ) . doubleValue ( ) ; }


signature:
public class BasicRollup { /** * merge rollups into this rollup . */
protected void computeFromRollups ( Points < BasicRollup > input ) throws IOException { } }

implementation:
computeFromRollupsHelper ( input ) ; if ( input . isEmpty ( ) ) { return ; } // See this and get mind blown :
// http : / / stackoverflow . com / questions / 18907262 / bounded - wildcard - related - compiler - error
Map < Long , ? extends Points . Point < ? extends BasicRollup > > points = input . getPoints ( ) ; for ( Map . Entry < Long , ? extends Points . Point < ? extends BasicRollup > > item : points . entrySet ( ) ) { BasicRollup rollup = item . getValue ( ) . getData ( ) ; if ( ! ( rollup instanceof BasicRollup ) ) { throw new IOException ( "Cannot create BasicRollup from type " + rollup . getClass ( ) . getName ( ) ) ; } BasicRollup basicRollup = ( BasicRollup ) rollup ; sum += basicRollup . getSum ( ) ; }


signature:
public class LocatorCache { /** * Checks if Locator is recently inserted in the batch layer
 * @ param loc
 * @ return */
public synchronized boolean isLocatorCurrentInBatchLayer ( Locator loc ) { } }

implementation: LocatorCacheEntry entry = insertedLocators . getIfPresent ( loc . toString ( ) ) ; return entry != null && entry . isBatchCurrent ( ) ;


signature:
public class LocatorCache { /** * Checks if Locator is recently inserted in the discovery layer
 * @ param loc
 * @ return */
public synchronized boolean isLocatorCurrentInDiscoveryLayer ( Locator loc ) { } }

implementation: LocatorCacheEntry entry = insertedLocators . getIfPresent ( loc . toString ( ) ) ; return entry != null && entry . isDiscoveryCurrent ( ) ;


signature:
public class LocatorCache { /** * Checks if Locator is recently inserted in the token discovery layer
 * @ param loc
 * @ return */
public synchronized boolean isLocatorCurrentInTokenDiscoveryLayer ( Locator loc ) { } }

implementation: LocatorCacheEntry entry = insertedLocators . getIfPresent ( loc . toString ( ) ) ; return entry != null && entry . isTokenDiscoveryCurrent ( ) ;


signature:
public class LocatorCache { /** * Check if the delayed locator is recently inserted for a given slot
 * @ param slot
 * @ param locator
 * @ return */
public synchronized boolean isDelayedLocatorForASlotCurrent ( int slot , Locator locator ) { } }

implementation: return insertedDelayedLocators . getIfPresent ( getLocatorSlotKey ( slot , locator ) ) != null ;


signature:
public class LocatorCache { /** * Marks the delayed locator as recently inserted for a given slot
 * @ param slot
 * @ param locator */
public synchronized void setDelayedLocatorForASlotCurrent ( int slot , Locator locator ) { } }

implementation: insertedDelayedLocators . put ( getLocatorSlotKey ( slot , locator ) , Boolean . TRUE ) ;


signature:
public class DSetIO { /** * Provides a way for the sub class to get a { @ link java . nio . ByteBuffer }
 * representation of a certain Rollup object .
 * @ param value
 * @ return */
@ Override protected ByteBuffer toByteBuffer ( Object value ) { } }

implementation: if ( ! ( value instanceof BluefloodSetRollup ) ) { throw new IllegalArgumentException ( "toByteBuffer(): expecting BluefloodSetRollup class but got " + value . getClass ( ) . getSimpleName ( ) ) ; } BluefloodSetRollup setRollup = ( BluefloodSetRollup ) value ; return serDes . serialize ( setRollup ) ;


signature:
public class Range { /** * given a start and stop time , return an iterator over ranges in * this * granularity that should be rolled up into
 * single points in the next coarser granularity .
 * Here is an example : Given start / end ( s , e ) , we need to return all the ranges in Y that correspond to rollup
 * periods A , B , C . This involves returning ranges prior to ( s ) and after ( e ) that map to parts of ( A ) and ( C ) in
 * the coarser rollup .
 * X [ A | B | C | D ]
 * Y [ | | s | | | | e | | | | | ] \
 * @ param startMillis
 * @ param stopMillis
 * @ return */
public static Iterable < Range > getRangesToRollup ( Granularity g , final long startMillis , final long stopMillis ) throws GranularityException { } }

implementation: final long snappedStartMillis = g . coarser ( ) . snapMillis ( startMillis ) ; final long snappedStopMillis = g . coarser ( ) . snapMillis ( stopMillis + g . coarser ( ) . milliseconds ( ) ) ; return new IntervalRangeIterator ( g , snappedStartMillis , snappedStopMillis ) ;


signature:
public class Range { /** * Returns a mapping of ranges in the coarser granularity to the sub - ranges in finer granularity
 * Here is an example : Given start / end ( s , e ) , we need to return mapping between ranges in Y that will be mapped to
 * a single range in X . From the example above , it will be mapping from A to all the sub - ranges in Y that get rolled
 * to a single point in A
 * @ param g Coarser Granularity
 * @ param range Range to be mapped
 * @ return
 * @ throws GranularityException */
public static Map < Range , Iterable < Range > > mapFinerRanges ( Granularity g , Range range ) throws GranularityException { } }

implementation: if ( range . getStart ( ) >= range . getStop ( ) ) throw new IllegalArgumentException ( "start cannot be greater than end. Start: " + range . getStart ( ) + " Stop:" + range . getStop ( ) ) ; final long snappedStartMillis = g . snapMillis ( range . getStart ( ) ) ; final long snappedStopMillis = g . snapMillis ( range . getStop ( ) + g . milliseconds ( ) ) ; HashMap < Range , Iterable < Range > > rangeMap = new HashMap < Range , Iterable < Range > > ( ) ; long tempStartMillis = snappedStartMillis ; int numberOfMillis = g . milliseconds ( ) ; while ( tempStartMillis <= ( snappedStopMillis - numberOfMillis ) ) { Range slotRange = new Range ( tempStartMillis , tempStartMillis + numberOfMillis ) ; rangeMap . put ( slotRange , new IntervalRangeIterator ( g . finer ( ) , slotRange . start , slotRange . stop ) ) ; tempStartMillis = tempStartMillis + numberOfMillis ; } return rangeMap ;


signature:
public class Range { /** * return the Ranges for an interval at this granularity
 * @ param from start time
 * @ param to end time
 * @ return Range [ ] */
public static Iterable < Range > rangesForInterval ( Granularity g , final long from , final long to ) { } }

implementation: if ( g == Granularity . FULL ) { return Arrays . asList ( new Range ( from , to ) ) ; } final long snappedStartMillis = g . snapMillis ( from ) ; final long snappedStopMillis = g . snapMillis ( to + g . milliseconds ( ) ) ; return new IntervalRangeIterator ( g , snappedStartMillis , snappedStopMillis ) ;


signature:
public class AstyanaxWriter { /** * single column updates ) . */
public void insertFull ( Collection < ? extends IMetric > metrics , boolean isRecordingDelayedMetrics , Clock clock ) throws ConnectionException { } }

implementation:
Timer . Context ctx = Instrumentation . getWriteTimerContext ( CassandraModel . CF_METRICS_FULL_NAME ) ; try { MutationBatch mutationBatch = keyspace . prepareMutationBatch ( ) ; for ( IMetric metric : metrics ) { final Locator locator = metric . getLocator ( ) ; // key = shard
// col = locator ( acct + entity + check + dimension . metric )
// value = < nothing >
if ( ! LocatorCache . getInstance ( ) . isLocatorCurrentInBatchLayer ( locator ) ) { if ( mutationBatch != null ) insertLocator ( locator , mutationBatch ) ; LocatorCache . getInstance ( ) . setLocatorCurrentInBatchLayer ( locator ) ; } if ( isRecordingDelayedMetrics ) { // retaining the same conditional logic that was used to insertLocator ( locator , batch ) above .
if ( mutationBatch != null ) { insertLocatorIfDelayed ( metric , mutationBatch , clock ) ; } } insertMetric ( metric , mutationBatch ) ; Instrumentation . markFullResMetricWritten ( ) ; } // insert it
try { mutationBatch . execute ( ) ; } catch ( ConnectionException e ) { Instrumentation . markWriteError ( e ) ; log . error ( "Connection exception during insertFull" , e ) ; throw e ; } } finally { ctx . stop ( ) ; }


signature:
public class AstyanaxWriter { /** * This method inserts the locator into the metric _ delayed _ locator column family , if the metric is delayed .
 * @ param metric
 * @ param mutationBatch
 * @ param clock */
private void insertLocatorIfDelayed ( IMetric metric , MutationBatch mutationBatch , Clock clock ) { } }

implementation:
Locator locator = metric . getLocator ( ) ; long delay = clock . now ( ) . getMillis ( ) - metric . getCollectionTime ( ) ; if ( delay > MAX_AGE_ALLOWED ) { // track locator for configured granularity level . to re - roll only the delayed locator ' s for that slot
int slot = DELAYED_METRICS_STORAGE_GRANULARITY . slot ( metric . getCollectionTime ( ) ) ; if ( ! LocatorCache . getInstance ( ) . isDelayedLocatorForASlotCurrent ( slot , locator ) ) { insertDelayedLocator ( DELAYED_METRICS_STORAGE_GRANULARITY , slot , locator , mutationBatch ) ; LocatorCache . getInstance ( ) . setDelayedLocatorForASlotCurrent ( slot , locator ) ; } }


signature:
public class AstyanaxWriter { /** * numeric only ! */
public final void insertLocator ( Locator locator , MutationBatch mutationBatch ) { } }

implementation: mutationBatch . withRow ( CassandraModel . CF_METRICS_LOCATOR , ( long ) Util . getShard ( locator . toString ( ) ) ) . putEmptyColumn ( locator , TenantTtlProvider . LOCATOR_TTL ) ;


signature:
public class AstyanaxWriter { /** * numeric only ! */
public final void insertDelayedLocator ( Granularity g , int slot , Locator locator , MutationBatch mutationBatch ) { } }

implementation: int shard = Util . getShard ( locator . toString ( ) ) ; mutationBatch . withRow ( CassandraModel . CF_METRICS_DELAYED_LOCATOR , SlotKey . of ( g , slot , shard ) ) . putEmptyColumn ( locator , TenantTtlProvider . DELAYED_LOCATOR_TTL ) ;


signature:
public class AstyanaxWriter { /** * generic IMetric insertion . All other metric insertion methods could use this one . */
public void insertMetrics ( Collection < IMetric > metrics , ColumnFamily cf , boolean isRecordingDelayedMetrics , Clock clock ) throws ConnectionException { } }

implementation:
Timer . Context ctx = Instrumentation . getWriteTimerContext ( cf . getName ( ) ) ; Multimap < Locator , IMetric > map = asMultimap ( metrics ) ; MutationBatch batch = keyspace . prepareMutationBatch ( ) ; try { for ( Locator locator : map . keySet ( ) ) { ColumnListMutation < Long > mutation = batch . withRow ( cf , locator ) ; for ( IMetric metric : map . get ( locator ) ) { mutation . putColumn ( metric . getCollectionTime ( ) , metric . getMetricValue ( ) , ( AbstractSerializer ) ( Serializers . serializerFor ( metric . getMetricValue ( ) . getClass ( ) ) ) , metric . getTtlInSeconds ( ) ) ; if ( cf . getName ( ) . equals ( CassandraModel . CF_METRICS_PREAGGREGATED_FULL_NAME ) ) { Instrumentation . markFullResPreaggregatedMetricWritten ( ) ; } if ( isRecordingDelayedMetrics ) { // retaining the same conditional logic that was used to perform insertLocator ( locator , batch ) .
insertLocatorIfDelayed ( metric , batch , clock ) ; } } if ( ! LocatorCache . getInstance ( ) . isLocatorCurrentInBatchLayer ( locator ) ) { insertLocator ( locator , batch ) ; LocatorCache . getInstance ( ) . setLocatorCurrentInBatchLayer ( locator ) ; } } try { batch . execute ( ) ; } catch ( ConnectionException e ) { Instrumentation . markWriteError ( e ) ; log . error ( "Connection exception persisting data" , e ) ; throw e ; } } finally { ctx . stop ( ) ; }


signature:
public class DAbstractMetricsRW { /** * This method inserts a collection of { @ link com . rackspacecloud . blueflood . service . SingleRollupWriteContext } objects
 * to the appropriate Cassandra column family .
 * It performs the inserts by executing an UNLOGGED BATCH statement .
 * @ param writeContexts
 * @ throws IOException */
@ Override public void insertRollups ( List < SingleRollupWriteContext > writeContexts ) { } }

implementation:
if ( writeContexts . size ( ) == 0 ) { return ; } Timer . Context ctx = Instrumentation . getWriteTimerContext ( writeContexts . get ( 0 ) . getDestinationCF ( ) . getName ( ) ) ; try { BatchStatement batch = new BatchStatement ( BatchStatement . Type . UNLOGGED ) ; for ( SingleRollupWriteContext writeContext : writeContexts ) { Rollup rollup = writeContext . getRollup ( ) ; Locator locator = writeContext . getLocator ( ) ; Granularity granularity = writeContext . getGranularity ( ) ; int ttl = getTtl ( locator , rollup . getRollupType ( ) , granularity ) ; // lookup the right writer
RollupType rollupType = writeContext . getRollup ( ) . getRollupType ( ) ; DAbstractMetricIO io = getIO ( rollupType . name ( ) . toLowerCase ( ) , granularity ) ; Statement statement = io . createStatement ( locator , writeContext . getTimestamp ( ) , rollup , writeContext . getGranularity ( ) , ttl ) ; batch . add ( statement ) ; } Session session = DatastaxIO . getSession ( ) ; session . execute ( batch ) ; } catch ( Exception ex ) { Instrumentation . markWriteError ( ) ; LOG . error ( String . format ( "error writing locator batch of size %s, granularity %s" , writeContexts . size ( ) , writeContexts . get ( 0 ) . getGranularity ( ) ) , ex ) ; } finally { ctx . stop ( ) ; }


signature:
public class DAbstractMetricsRW { /** * Fetches { @ link com . rackspacecloud . blueflood . outputs . formats . MetricData } objects for the
 * specified { @ link com . rackspacecloud . blueflood . types . Locator } and
 * { @ link com . rackspacecloud . blueflood . types . Range } from the specified column family
 * @ param locator
 * @ param range
 * @ param granularity
 * @ return */
@ Override public MetricData getDatapointsForRange ( final Locator locator , Range range , Granularity granularity ) { } }

implementation: Map < Locator , MetricData > result = getDatapointsForRange ( new ArrayList < Locator > ( ) { { add ( locator ) ; } } , range , granularity ) ; return result . get ( locator ) ;


signature:
public class DAbstractMetricsRW { /** * Fetches { @ link com . rackspacecloud . blueflood . outputs . formats . MetricData } objects for the
 * specified { @ link com . rackspacecloud . blueflood . types . Locator } and
 * { @ link com . rackspacecloud . blueflood . types . Range } from the specified column family
 * This is a helper method used to get basic metrics as well as preaggregated metrics . Hence , both granularity
 * and columnFamily are required .
 * @ param locators
 * @ param range
 * @ param columnFamily
 * @ param granularity
 * @ return */
public Map < Locator , MetricData > getDatapointsForRange ( List < Locator > locators , Range range , String columnFamily , Granularity granularity ) { } }

implementation:
Timer . Context ctx = Instrumentation . getReadTimerContext ( columnFamily ) ; try { MetadataCache metadataCache = MetadataCache . getInstance ( ) ; // in this loop , we will fire all the executeAsync ( ) of
// various select statements , the collect all of the
// ResultSetFutures
Map < Locator , List < ResultSetFuture > > locatorToFuturesMap = new HashMap < Locator , List < ResultSetFuture > > ( ) ; Map < Locator , DAbstractMetricIO > locatorIOMap = new HashMap < Locator , DAbstractMetricIO > ( ) ; for ( Locator locator : locators ) { try { String rType = metadataCache . get ( locator , MetricMetadata . ROLLUP_TYPE . name ( ) . toLowerCase ( ) ) ; DAbstractMetricIO io = getIO ( rType , granularity ) ; // put everything in a map of locator - > io so
// we can use em up later
locatorIOMap . put ( locator , io ) ; // do the query
List < ResultSetFuture > selectFutures = io . selectForLocatorAndRange ( columnFamily , locator , range ) ; // add all ResultSetFutures for a particular locator together
List < ResultSetFuture > existing = locatorToFuturesMap . get ( locator ) ; if ( existing == null ) { existing = new ArrayList < ResultSetFuture > ( ) ; locatorToFuturesMap . put ( locator , existing ) ; } existing . addAll ( selectFutures ) ; } catch ( CacheException ex ) { Instrumentation . markReadError ( ) ; LOG . error ( String . format ( "Error looking up locator %s in cache" , locator ) , ex ) ; } } return resultSetsToMetricData ( locatorToFuturesMap , locatorIOMap , columnFamily , range ) ; } finally { ctx . stop ( ) ; }


signature:
public class DAbstractMetricsRW { /** * Fetches a { @ link com . rackspacecloud . blueflood . types . Points } object for a
 * particular locator and rollupType from the specified column family and
 * range
 * @ param locator
 * @ param rollupType
 * @ param range
 * @ param columnFamilyName
 * @ return */
@ Override public Points getDataToRollup ( final Locator locator , RollupType rollupType , Range range , String columnFamilyName ) throws IOException { } }

implementation:
Timer . Context ctx = Instrumentation . getReadTimerContext ( columnFamilyName ) ; try { // read the rollup object from the proper IO class
DAbstractMetricIO io = getIO ( rollupType . name ( ) . toLowerCase ( ) , CassandraModel . getGranularity ( columnFamilyName ) ) ; Table < Locator , Long , Object > locatorTimestampRollup = io . getRollupsForLocator ( locator , columnFamilyName , range ) ; Points points = new Points ( ) ; for ( Table . Cell < Locator , Long , Object > cell : locatorTimestampRollup . cellSet ( ) ) { points . add ( createPoint ( cell . getColumnKey ( ) , cell . getValue ( ) ) ) ; } return points ; } catch ( Exception e ) { Instrumentation . markReadError ( ) ; LOG . error ( String . format ( "Unable to read locator=%s rolluptype=%s columnFamilyName=%s for rollup" , locator , rollupType . name ( ) , columnFamilyName ) , e ) ; throw new IOException ( e ) ; } finally { ctx . stop ( ) ; }


signature:
public class DAbstractMetricsRW { /** * Converts a list of { @ link com . datastax . driver . core . ResultSetFuture } for each
 * { @ link com . rackspacecloud . blueflood . types . Locator } to
 * { @ link com . rackspacecloud . blueflood . outputs . formats . MetricData } object .
 * @ param resultSets
 * @ param locatorIO
 * @ param columnFamily
 * @ return */
protected Map < Locator , MetricData > resultSetsToMetricData ( Map < Locator , List < ResultSetFuture > > resultSets , Map < Locator , DAbstractMetricIO > locatorIO , String columnFamily , Range range ) { } }

implementation:
MetadataCache metadataCache = MetadataCache . getInstance ( ) ; // iterate through all ResultSetFuture
Map < Locator , MetricData > locatorMetricDataMap = new HashMap < Locator , MetricData > ( ) ; for ( Map . Entry < Locator , List < ResultSetFuture > > entry : resultSets . entrySet ( ) ) { Locator locator = entry . getKey ( ) ; List < ResultSetFuture > futures = entry . getValue ( ) ; DAbstractMetricIO io = locatorIO . get ( locator ) ; // get ResultSets to a Table of locator , timestamp , rollup
Table < Locator , Long , Object > locatorTimestampRollup = io . toLocatorTimestampValue ( futures , locator , columnFamily , range ) ; Map < Long , Object > tsRollupMap = locatorTimestampRollup . row ( locator ) ; // convert to Points and MetricData
Points points = convertToPoints ( tsRollupMap ) ; // create MetricData
MetricData metricData = new MetricData ( points , metadataCache . getUnitString ( locator ) ) ; locatorMetricDataMap . put ( locator , metricData ) ; } return locatorMetricDataMap ;


signature:
public class DAbstractMetricsRW { /** * Given a map of timestamps - > Rollups / SimpleNumbers , return RollupType , or null , if not a Rollup .
 * @ param tsRollupMap
 * @ return */
private RollupType getRollupType ( Map < Long , Object > tsRollupMap ) { } }

implementation: if ( tsRollupMap . isEmpty ( ) ) return null ; else { Object value = tsRollupMap . values ( ) . iterator ( ) . next ( ) ; return value instanceof Rollup ? ( ( Rollup ) value ) . getRollupType ( ) : null ; }


signature:
public class DAbstractMetricsRW { /** * Determines if the metric is considered delayed or not
 * @ param metric
 * @ return */
protected boolean isDelayed ( IMetric metric ) { } }

implementation: long delay = clock . now ( ) . getMillis ( ) - metric . getCollectionTime ( ) ; return delay > MAX_AGE_ALLOWED ;


signature:
public class DAbstractMetricsRW { /** * This method inserts the locator into the metric _ delayed _ locator column family , if the metric is delayed .
 * @ param metric
 * @ throws IOException */
protected void insertLocatorIfDelayed ( IMetric metric ) throws IOException { } }

implementation: Locator locator = metric . getLocator ( ) ; if ( isDelayed ( metric ) ) { int slot = getDelayedSlot ( metric ) ; if ( ! LocatorCache . getInstance ( ) . isDelayedLocatorForASlotCurrent ( slot , locator ) ) { delayedLocatorIO . insertLocator ( DELAYED_METRICS_STORAGE_GRANULARITY , slot , locator ) ; LocatorCache . getInstance ( ) . setDelayedLocatorForASlotCurrent ( slot , locator ) ; } }


signature:
public class DAbstractMetricsRW { /** * Returns a BoundStatement if a metric needs to be inserted to the metrics _ delayed _ locator
 * Column Family . Returns null otherwise .
 * @ param metric
 * @ return */
protected BoundStatement getBoundStatementForMetricIfDelayed ( IMetric metric ) { } }

implementation: Locator locator = metric . getLocator ( ) ; if ( isDelayed ( metric ) ) { int slot = getDelayedSlot ( metric ) ; if ( ! LocatorCache . getInstance ( ) . isDelayedLocatorForASlotCurrent ( slot , locator ) ) { LocatorCache . getInstance ( ) . setDelayedLocatorForASlotCurrent ( slot , locator ) ; return delayedLocatorIO . getBoundStatementForLocator ( DELAYED_METRICS_STORAGE_GRANULARITY , slot , locator ) ; } } return null ;


signature:
public class ElasticTokensIO { /** * This method queries elasticsearch for a given glob query and returns list of { @ link MetricName } ' s .
 * @ param tenantId
 * @ param query is glob
 * @ param indexes
 * @ return */
private List < MetricName > searchESByIndexes ( String tenantId , String query , String [ ] indexes ) { } }

implementation: if ( StringUtils . isEmpty ( query ) ) return new ArrayList < > ( ) ; Timer . Context timerCtx = esMetricNamesQueryTimer . time ( ) ; try { String response = elasticsearchRestHelper . fetchTokenDocuments ( indexes , tenantId , query ) ; List < MetricName > metricNames = getMetricNames ( response ) ; Set < MetricName > uniqueMetricNames = new HashSet < > ( metricNames ) ; return new ArrayList < > ( uniqueMetricNames ) ; } catch ( IOException e ) { log . error ( "IOException: Elasticsearch token query failed for tenantId {}. {}" , tenantId , e . getMessage ( ) ) ; throw new RuntimeException ( String . format ( "searchESByIndexes failed with message: %s" , e . getMessage ( ) ) , e ) ; } catch ( Exception e ) { log . error ( "Exception: Elasticsearch token query failed for tenantId {}. {}" , tenantId , e . getMessage ( ) ) ; throw new RuntimeException ( String . format ( "searchESByIndexes failed with message: %s" , e . getMessage ( ) ) , e ) ; } finally { timerCtx . stop ( ) ; }


signature:
public class AstyanaxReader { /** * Method that returns all metadata for a given locator as a map .
 * @ param locator locator name
 * @ return Map of metadata for that locator
 * @ throws RuntimeException ( com . netflix . astyanax . connectionpool . exceptions . ConnectionException ) */
public Map < String , String > getMetadataValues ( Locator locator ) { } }

implementation: Timer . Context ctx = Instrumentation . getReadTimerContext ( CassandraModel . CF_METRICS_METADATA_NAME ) ; try { final ColumnList < String > results = keyspace . prepareQuery ( CassandraModel . CF_METRICS_METADATA ) . getKey ( locator ) . execute ( ) . getResult ( ) ; return new HashMap < String , String > ( ) { { for ( Column < String > result : results ) { put ( result . getName ( ) , result . getValue ( StringMetadataSerializer . get ( ) ) ) ; } } } ; } catch ( NotFoundException ex ) { Instrumentation . markNotFound ( CassandraModel . CF_METRICS_METADATA_NAME ) ; return null ; } catch ( ConnectionException e ) { log . error ( "Error reading metadata value" , e ) ; Instrumentation . markReadError ( e ) ; throw new RuntimeException ( e ) ; } finally { ctx . stop ( ) ; }


signature:
public class AstyanaxReader { /** * todo : A better interface may be to pass the serializer in instead of the class type . */
public < T extends Rollup > Points < T > getDataToRoll ( Class < T > type , final Locator locator , Range range , ColumnFamily < Locator , Long > cf ) throws IOException { } }

implementation:
AbstractSerializer serializer = Serializers . serializerFor ( type ) ; // special cases . : ( the problem here is that the normal full res serializer returns Number instances instead of
// SimpleNumber instances .
// todo : this logic will only become more complicated . It needs to be in its own method and the serializer needs
// to be known before we ever get to this method ( see above comment ) .
if ( cf == CassandraModel . CF_METRICS_FULL ) { serializer = Serializers . simpleNumberSerializer ; } else if ( cf == CassandraModel . CF_METRICS_PREAGGREGATED_FULL ) { // consider a method for this . getSerializer ( CF , TYPE ) ;
if ( type . equals ( BluefloodTimerRollup . class ) ) { serializer = Serializers . timerRollupInstance ; } else if ( type . equals ( BluefloodSetRollup . class ) ) { serializer = Serializers . setRollupInstance ; } else if ( type . equals ( BluefloodGaugeRollup . class ) ) { serializer = Serializers . gaugeRollupInstance ; } else if ( type . equals ( BluefloodCounterRollup . class ) ) { serializer = Serializers . counterRollupInstance ; } else { serializer = Serializers . simpleNumberSerializer ; } } ColumnList < Long > cols = getColumnsFromDB ( locator , cf , range ) ; Points < T > points = new Points < T > ( ) ; try { for ( Column < Long > col : cols ) { points . add ( new Points . Point < T > ( col . getName ( ) , ( T ) col . getValue ( serializer ) ) ) ; } // we only want to count the number of points we
// get when we ' re querying the metrics _ full
// we don ' t do for aggregated or other granularities
if ( cf == CassandraModel . CF_METRICS_FULL ) { Instrumentation . getRawPointsIn5MinHistogram ( ) . update ( points . getPoints ( ) . size ( ) ) ; } } catch ( RuntimeException ex ) { log . error ( "Problem deserializing data for " + locator + " (" + range + ") from " + cf . getName ( ) , ex ) ; throw new IOException ( ex ) ; } return points ;


signature:
public class AstyanaxReader { /** * Get data points for a particular { @ link Locator } , { @ link Range } , and
 * { @ link Granularity } .
 * This method is eventually called from Rollup process .
 * @ param locator
 * @ param range
 * @ param gran
 * @ return a { @ link MetricData } containing the data points requested */
public MetricData getDatapointsForRange ( Locator locator , Range range , Granularity gran ) { } }

implementation: RollupType rollupType = RollupType . BF_BASIC ; String rollupTypeStr = metaCache . safeGet ( locator , rollupTypeCacheKey ) ; if ( rollupTypeStr != null ) { rollupType = RollupType . fromString ( rollupTypeStr ) ; } return getNumericMetricDataForRange ( locator , range , gran , rollupType ) ;


signature:
public class AstyanaxReader { /** * Get data points for multiple { @ link Locator } , for the specified { @ link Range } and
 * { @ link Granularity } .
 * This method is eventually called from all the output handlers for query requests .
 * @ param locators
 * @ param range
 * @ param gran
 * @ return */
public Map < Locator , MetricData > getDatapointsForRange ( List < Locator > locators , Range range , Granularity gran ) { } }

implementation:
ListMultimap < ColumnFamily , Locator > locatorsByCF = ArrayListMultimap . create ( ) ; Map < Locator , MetricData > results = new HashMap < Locator , MetricData > ( ) ; for ( Locator locator : locators ) { try { RollupType rollupType = RollupType . fromString ( metaCache . get ( locator , MetricMetadata . ROLLUP_TYPE . name ( ) . toLowerCase ( ) ) ) ; ColumnFamily cf = CassandraModel . getColumnFamily ( rollupType , gran ) ; List < Locator > locs = locatorsByCF . get ( cf ) ; locs . add ( locator ) ; } catch ( Exception e ) { // pass for now . need metric to figure this stuff out .
log . error ( String . format ( "error getting datapoints for locator %s, range %s, granularity %s" , locator , range . toString ( ) , gran . toString ( ) ) , e ) ; } } for ( ColumnFamily CF : locatorsByCF . keySet ( ) ) { List < Locator > locs = locatorsByCF . get ( CF ) ; results . putAll ( getNumericDataForRangeLocatorList ( range , gran , CF , locs ) ) ; } return results ;


signature:
public class ABasicMetricsRW { /** * Fetches a { @ link com . rackspacecloud . blueflood . types . Points } object for a
 * particular locator and rollupType from the specified column family and
 * range
 * @ param locator
 * @ param rollupType
 * @ param range
 * @ param columnFamilyName
 * @ return */
@ Override public Points < BasicRollup > getDataToRollup ( final Locator locator , RollupType rollupType , Range range , String columnFamilyName ) throws IOException { } }

implementation: return AstyanaxReader . getInstance ( ) . getDataToRoll ( BasicRollup . class , locator , range , CassandraModel . getColumnFamily ( columnFamilyName ) ) ;


signature:
public class Serializers { /** * return a serializer for a specific type */
public static < T > AbstractSerializer < T > serializerFor ( Class < T > type ) { } }

implementation: if ( type == null ) throw new RuntimeException ( "serializable type cannot be null" , new SerializationException ( "serializable type cannot be null" ) ) ; else if ( type . equals ( String . class ) ) throw new RuntimeException ( "We don't serialize strings anymore" , new SerializationException ( "We don't serialize strings anymore" ) ) ; if ( type . equals ( BasicRollup . class ) ) return ( AbstractSerializer < T > ) basicRollupInstance ; else if ( type . equals ( BluefloodTimerRollup . class ) ) return ( AbstractSerializer < T > ) timerRollupInstance ; else if ( type . equals ( BluefloodCounterRollup . class ) ) return ( AbstractSerializer < T > ) counterRollupInstance ; else if ( type . equals ( BluefloodGaugeRollup . class ) ) return ( AbstractSerializer < T > ) gaugeRollupInstance ; else if ( type . equals ( BluefloodSetRollup . class ) ) return ( AbstractSerializer < T > ) setRollupInstance ; else if ( type . equals ( SimpleNumber . class ) ) return ( AbstractSerializer < T > ) fullInstance ; else if ( type . equals ( Integer . class ) ) return ( AbstractSerializer < T > ) fullInstance ; else if ( type . equals ( Long . class ) ) return ( AbstractSerializer < T > ) fullInstance ; else if ( type . equals ( Double . class ) ) return ( AbstractSerializer < T > ) fullInstance ; else if ( type . equals ( Float . class ) ) return ( AbstractSerializer < T > ) fullInstance ; else if ( type . equals ( byte [ ] . class ) ) return ( AbstractSerializer < T > ) fullInstance ; else if ( type . equals ( Object . class ) ) return ( AbstractSerializer < T > ) fullInstance ; else return ( AbstractSerializer < T > ) fullInstance ;


signature:
public class MediaTypeChecker { /** * Checks the Content - Type header to see if clients specify the right
 * media type
 * @ param headers
 * @ return */
public boolean isContentTypeValid ( HttpHeaders headers ) { } }

implementation:
String contentType = headers . get ( HttpHeaders . Names . CONTENT_TYPE ) ; // if we get no Content - Type or we get application / json , then it ' s valid
// any other , it ' s invalid
return ( Strings . isNullOrEmpty ( contentType ) || contentType . toLowerCase ( ) . contains ( MEDIA_TYPE_APPLICATION_JSON ) ) ;


signature:
public class MediaTypeChecker { /** * Checks the Accept header to see if clients accept the correct
 * media type
 * @ param headers
 * @ return */
public boolean isAcceptValid ( HttpHeaders headers ) { } }

implementation:
String accept = headers . get ( HttpHeaders . Names . ACCEPT ) ; // if we get no Accept ( which means * / * ) , or * / * ,
// or application / json , then it ' s valid
return ( Strings . isNullOrEmpty ( accept ) || accept . contains ( ACCEPT_ALL ) || accept . toLowerCase ( ) . contains ( MEDIA_TYPE_APPLICATION_JSON ) ) ;