blueflood-core/src/main/java/com/rackspacecloud/blueflood/io/astyanax/AstyanaxWriter.java
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();
        }
    }
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();
        }
    }
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.
generic IMetric insertion . All other metric insertion methods could use this one .