public void insertMetrics ( Collection < IMetric > metrics , ColumnFamily cf , boolean isRecordingDelayedMetrics , Clock clock ) throws ConnectionException { 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 ( ) ; } }
generic IMetric insertion . All other metric insertion methods could use this one .