metron-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nickal...@apache.org
Subject [20/21] metron git commit: Merge remote-tracking branch 'apache/master' into feature/METRON-1699-create-batch-profiler
Date Tue, 18 Sep 2018 14:55:01 GMT
http://git-wip-us.apache.org/repos/asf/metron/blob/cad2f408/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/ProfileBuilderBoltTest.java
----------------------------------------------------------------------
diff --cc metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/ProfileBuilderBoltTest.java
index fc94afa,0000000..3f16edd
mode 100644,000000..100644
--- a/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/ProfileBuilderBoltTest.java
+++ b/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/ProfileBuilderBoltTest.java
@@@ -1,356 -1,0 +1,378 @@@
 +/*
 + *
 + *  Licensed to the Apache Software Foundation (ASF) under one
 + *  or more contributor license agreements.  See the NOTICE file
 + *  distributed with this work for additional information
 + *  regarding copyright ownership.  The ASF licenses this file
 + *  to you under the Apache License, Version 2.0 (the
 + *  "License"); you may not use this file except in compliance
 + *  with the License.  You may obtain a copy of the License at
 + *
 + *      http://www.apache.org/licenses/LICENSE-2.0
 + *
 + *  Unless required by applicable law or agreed to in writing, software
 + *  distributed under the License is distributed on an "AS IS" BASIS,
 + *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 + *  See the License for the specific language governing permissions and
 + *  limitations under the License.
 + *
 + */
 +
 +package org.apache.metron.profiler.storm;
 +
 +import org.apache.metron.common.configuration.profiler.ProfileConfig;
 +import org.apache.metron.common.configuration.profiler.ProfilerConfigurations;
 +import org.apache.metron.profiler.MessageDistributor;
 +import org.apache.metron.profiler.MessageRoute;
 +import org.apache.metron.profiler.ProfileMeasurement;
 +import org.apache.metron.profiler.storm.integration.MessageBuilder;
 +import org.apache.metron.test.bolt.BaseBoltTest;
 +import org.apache.storm.task.OutputCollector;
 +import org.apache.storm.topology.OutputFieldsDeclarer;
 +import org.apache.storm.topology.base.BaseWindowedBolt;
 +import org.apache.storm.tuple.Fields;
 +import org.apache.storm.tuple.Tuple;
 +import org.apache.storm.tuple.Values;
 +import org.apache.storm.windowing.TupleWindow;
 +import org.json.simple.JSONObject;
 +import org.junit.Before;
 +import org.junit.Test;
 +import org.mockito.ArgumentCaptor;
 +
 +import java.io.IOException;
 +import java.util.Arrays;
 +import java.util.Collections;
 +import java.util.HashMap;
 +import java.util.List;
 +import java.util.concurrent.TimeUnit;
 +import java.util.stream.Collectors;
 +
 +import static org.junit.Assert.assertEquals;
 +import static org.mockito.Matchers.any;
 +import static org.mockito.Matchers.eq;
++import static org.mockito.Mockito.doThrow;
 +import static org.mockito.Mockito.mock;
 +import static org.mockito.Mockito.times;
 +import static org.mockito.Mockito.verify;
 +import static org.mockito.Mockito.when;
 +
 +/**
 + * Tests the ProfileBuilderBolt.
 + */
 +public class ProfileBuilderBoltTest extends BaseBoltTest {
 +
 +  private JSONObject message1;
 +  private JSONObject message2;
 +  private ProfileConfig profile1;
 +  private ProfileConfig profile2;
 +  private ProfileMeasurementEmitter emitter;
 +  private ManualFlushSignal flushSignal;
 +  private ProfileMeasurement measurement;
 +
 +  @Before
 +  public void setup() throws Exception {
 +
 +    message1 = new MessageBuilder()
 +            .withField("ip_src_addr", "10.0.0.1")
 +            .withField("value", "22")
 +            .build();
 +
 +    message2 = new MessageBuilder()
 +            .withField("ip_src_addr", "10.0.0.2")
 +            .withField("value", "22")
 +            .build();
 +
 +    profile1 = new ProfileConfig()
 +            .withProfile("profile1")
 +            .withForeach("ip_src_addr")
 +            .withInit("x", "0")
 +            .withUpdate("x", "x + 1")
 +            .withResult("x");
 +
 +    profile2 = new ProfileConfig()
 +            .withProfile("profile2")
 +            .withForeach("ip_src_addr")
 +            .withInit(Collections.singletonMap("x", "0"))
 +            .withUpdate(Collections.singletonMap("x", "x + 1"))
 +            .withResult("x");
 +
 +    measurement = new ProfileMeasurement()
 +            .withEntity("entity1")
 +            .withProfileName("profile1")
 +            .withPeriod(1000, 500, TimeUnit.MILLISECONDS)
 +            .withProfileValue(22);
 +
 +    flushSignal = new ManualFlushSignal();
 +    flushSignal.setFlushNow(false);
 +  }
 +
 +  /**
 +   * The bolt should extract a message and timestamp from a tuple and
 +   * pass that to a {@code MessageDistributor}.
 +   */
 +  @Test
 +  public void testExtractMessage() throws Exception {
 +
 +    ProfileBuilderBolt bolt = createBolt();
 +
 +    // create a mock
 +    MessageDistributor distributor = mock(MessageDistributor.class);
 +    bolt.withMessageDistributor(distributor);
 +
 +    // create a tuple
 +    final long timestamp1 = 100000000L;
 +    Tuple tuple1 = createTuple("entity1", message1, profile1, timestamp1);
 +
 +    // execute the bolt
 +    TupleWindow tupleWindow = createWindow(tuple1);
 +    bolt.execute(tupleWindow);
 +
 +    // the message should have been extracted from the tuple and passed to the MessageDistributor
 +    verify(distributor).distribute(any(MessageRoute.class), any());
 +  }
 +
 +
 +  /**
 +   * If the {@code FlushSignal} tells the bolt to flush, it should flush the {@code MessageDistributor}
 +   * and emit the {@code ProfileMeasurement} values from all active profiles.
 +   */
 +  @Test
 +  public void testFlushActiveProfiles() throws Exception {
 +
 +    ProfileBuilderBolt bolt = createBolt();
 +
 +    // create a mock that returns the profile measurement above
 +    MessageDistributor distributor = mock(MessageDistributor.class);
 +    when(distributor.flush()).thenReturn(Collections.singletonList(measurement));
 +    bolt.withMessageDistributor(distributor);
 +
 +    // signal the bolt to flush
 +    flushSignal.setFlushNow(true);
 +
 +    // execute the bolt
 +    Tuple tuple1 = createTuple("entity1", message1, profile1, 1000L);
 +    TupleWindow tupleWindow = createWindow(tuple1);
 +    bolt.execute(tupleWindow);
 +
 +    // a profile measurement should be emitted by the bolt
 +    List<ProfileMeasurement> measurements = getProfileMeasurements(outputCollector,
1);
 +    assertEquals(1, measurements.size());
 +    assertEquals(measurement, measurements.get(0));
 +  }
 +
 +  /**
 +   * If the {@code FlushSignal} tells the bolt NOT to flush, nothing should be emitted.
 +   */
 +  @Test
 +  public void testDoNotFlushActiveProfiles() throws Exception {
 +
 +    ProfileBuilderBolt bolt = createBolt();
 +
 +    // create a mock where flush() returns the profile measurement above
 +    MessageDistributor distributor = mock(MessageDistributor.class);
 +    when(distributor.flush()).thenReturn(Collections.singletonList(measurement));
 +    bolt.withMessageDistributor(distributor);
 +
 +    // there is no flush signal
 +    flushSignal.setFlushNow(false);
 +
 +    // execute the bolt
 +    Tuple tuple1 = createTuple("entity1", message1, profile1, 1000L);
 +    TupleWindow tupleWindow = createWindow(tuple1);
 +    bolt.execute(tupleWindow);
 +
 +    // nothing should have been emitted
 +    getProfileMeasurements(outputCollector, 0);
 +  }
 +
 +  /**
 +   * Expired profiles should be flushed regularly, even if no input telemetry
 +   * has been received.
 +   */
 +  @Test
 +  public void testFlushExpiredProfiles() throws Exception {
 +
 +    ProfileBuilderBolt bolt = createBolt();
 +
 +    // create a mock where flushExpired() returns the profile measurement above
 +    MessageDistributor distributor = mock(MessageDistributor.class);
 +    when(distributor.flushExpired()).thenReturn(Collections.singletonList(measurement));
 +    bolt.withMessageDistributor(distributor);
 +
 +    // execute test by flushing expired profiles. this is normally triggered by a timer
task.
 +    bolt.flushExpired();
 +
 +    // a profile measurement should be emitted by the bolt
 +    List<ProfileMeasurement> measurements = getProfileMeasurements(outputCollector,
1);
 +    assertEquals(1, measurements.size());
 +    assertEquals(measurement, measurements.get(0));
 +  }
 +
 +  /**
 +   * A {@link ProfileMeasurement} is built for each profile/entity pair.  The measurement
should be emitted to each
 +   * destination defined by the profile. By default, a profile uses both Kafka and HBase
as destinations.
 +   */
 +  @Test
 +  public void testEmitters() throws Exception {
 +
 +    // defines the zk configurations accessible from the bolt
 +    ProfilerConfigurations configurations = new ProfilerConfigurations();
 +    configurations.updateGlobalConfig(Collections.emptyMap());
 +
 +    // create the bolt with 3 destinations
 +    ProfileBuilderBolt bolt = (ProfileBuilderBolt) new ProfileBuilderBolt()
 +            .withProfileTimeToLive(30, TimeUnit.MINUTES)
 +            .withPeriodDuration(10, TimeUnit.MINUTES)
 +            .withMaxNumberOfRoutes(Long.MAX_VALUE)
 +            .withZookeeperClient(client)
 +            .withZookeeperCache(cache)
 +            .withEmitter(new TestEmitter("destination1"))
 +            .withEmitter(new TestEmitter("destination2"))
 +            .withEmitter(new TestEmitter("destination3"))
 +            .withProfilerConfigurations(configurations)
 +            .withTumblingWindow(new BaseWindowedBolt.Duration(10, TimeUnit.MINUTES));
 +    bolt.prepare(new HashMap<>(), topologyContext, outputCollector);
 +
 +    // signal the bolt to flush
 +    bolt.withFlushSignal(flushSignal);
 +    flushSignal.setFlushNow(true);
 +
 +    // execute the bolt
 +    Tuple tuple1 = createTuple("entity", message1, profile1, System.currentTimeMillis());
 +    TupleWindow window = createWindow(tuple1);
 +    bolt.execute(window);
 +
 +    // validate measurements emitted to each
 +    verify(outputCollector, times(1)).emit(eq("destination1"), any());
 +    verify(outputCollector, times(1)).emit(eq("destination2"), any());
 +    verify(outputCollector, times(1)).emit(eq("destination3"), any());
 +  }
 +
++  @Test
++  public void testExceptionWhenFlushingExpiredProfiles() throws Exception {
++    // create an emitter that will throw an exception when emit() called
++    ProfileMeasurementEmitter badEmitter = mock(ProfileMeasurementEmitter.class);
++    doThrow(new RuntimeException("flushExpired() should catch this exception"))
++            .when(badEmitter)
++            .emit(any(), any());
++
++    // create a distributor that will return a profile measurement
++    MessageDistributor distributor = mock(MessageDistributor.class);
++    when(distributor.flushExpired()).thenReturn(Collections.singletonList(measurement));
++
++    // the bolt will use the bad emitter when flushExpired() is called
++    ProfileBuilderBolt bolt = (ProfileBuilderBolt) new ProfileBuilderBolt()
++            .withEmitter(badEmitter)
++            .withMessageDistributor(distributor);
++
++    // the exception thrown by the emitter should not bubble up
++    bolt.flushExpired();
++  }
++
 +  /**
 +   * Retrieves the ProfileMeasurement(s) (if any) that have been emitted.
 +   *
 +   * @param collector The Storm output collector.
 +   * @param expected The number of measurements expected.
 +   * @return A list of ProfileMeasurement(s).
 +   */
 +  private List<ProfileMeasurement> getProfileMeasurements(OutputCollector collector,
int expected) {
 +
 +    // the 'streamId' is defined by the DestinationHandler being used by the bolt
 +    final String streamId = emitter.getStreamId();
 +
 +    // capture the emitted tuple(s)
 +    ArgumentCaptor<Values> argCaptor = ArgumentCaptor.forClass(Values.class);
 +    verify(collector, times(expected))
 +            .emit(eq(streamId), argCaptor.capture());
 +
 +    // return the profile measurements that were emitted
 +    return argCaptor.getAllValues()
 +            .stream()
 +            .map(val -> (ProfileMeasurement) val.get(0))
 +            .collect(Collectors.toList());
 +  }
 +
 +  /**
 +   * Create a tuple that will contain the message, the entity name, and profile definition.
 +   * @param entity The entity name
 +   * @param message The telemetry message.
 +   * @param profile The profile definition.
 +   */
 +  private Tuple createTuple(String entity, JSONObject message, ProfileConfig profile, long
timestamp) {
 +
 +    Tuple tuple = mock(Tuple.class);
 +    when(tuple.getValueByField(eq(ProfileSplitterBolt.MESSAGE_TUPLE_FIELD))).thenReturn(message);
 +    when(tuple.getValueByField(eq(ProfileSplitterBolt.TIMESTAMP_TUPLE_FIELD))).thenReturn(timestamp);
 +    when(tuple.getValueByField(eq(ProfileSplitterBolt.ENTITY_TUPLE_FIELD))).thenReturn(entity);
 +    when(tuple.getValueByField(eq(ProfileSplitterBolt.PROFILE_TUPLE_FIELD))).thenReturn(profile);
 +
 +    return tuple;
 +  }
 +
 +  /**
 +   * Create a ProfileBuilderBolt to test.
 +   * @return A {@link ProfileBuilderBolt} to test.
 +   */
 +  private ProfileBuilderBolt createBolt() throws IOException {
 +
 +    // defines the zk configurations accessible from the bolt
 +    ProfilerConfigurations configurations = new ProfilerConfigurations();
 +    configurations.updateGlobalConfig(Collections.emptyMap());
 +
 +    emitter = new HBaseEmitter();
 +    ProfileBuilderBolt bolt = (ProfileBuilderBolt) new ProfileBuilderBolt()
 +            .withProfileTimeToLive(30, TimeUnit.MINUTES)
 +            .withMaxNumberOfRoutes(Long.MAX_VALUE)
 +            .withZookeeperClient(client)
 +            .withZookeeperCache(cache)
 +            .withEmitter(emitter)
 +            .withProfilerConfigurations(configurations)
 +            .withPeriodDuration(1, TimeUnit.MINUTES)
 +            .withTumblingWindow(new BaseWindowedBolt.Duration(30, TimeUnit.SECONDS));
 +    bolt.prepare(new HashMap<>(), topologyContext, outputCollector);
 +
 +    // set the flush signal AFTER calling 'prepare'
 +    bolt.withFlushSignal(flushSignal);
 +
 +    return bolt;
 +  }
 +
 +  /**
 +   * Creates a mock TupleWindow containing multiple tuples.
 +   * @param tuples The tuples to add to the window.
 +   */
 +  private TupleWindow createWindow(Tuple... tuples) {
 +
 +    TupleWindow window = mock(TupleWindow.class);
 +    when(window.get()).thenReturn(Arrays.asList(tuples));
 +    return window;
 +  }
 +
 +  /**
 +   * An implementation for testing purposes only.
 +   */
 +  private class TestEmitter implements ProfileMeasurementEmitter {
 +
 +    private String streamId;
 +
 +    public TestEmitter(String streamId) {
 +      this.streamId = streamId;
 +    }
 +
 +    @Override
 +    public String getStreamId() {
 +      return streamId;
 +    }
 +
 +    @Override
 +    public void declareOutputFields(OutputFieldsDeclarer declarer) {
 +      declarer.declareStream(getStreamId(), new Fields("measurement"));
 +    }
 +
 +    @Override
 +    public void emit(ProfileMeasurement measurement, OutputCollector collector) {
 +      collector.emit(getStreamId(), new Values(measurement));
 +    }
 +  }
 +}

http://git-wip-us.apache.org/repos/asf/metron/blob/cad2f408/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/integration/ProfilerIntegrationTest.java
----------------------------------------------------------------------
diff --cc metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/integration/ProfilerIntegrationTest.java
index 182600a,0000000..4389d42
mode 100644,000000..100644
--- a/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/integration/ProfilerIntegrationTest.java
+++ b/metron-analytics/metron-profiler-storm/src/test/java/org/apache/metron/profiler/storm/integration/ProfilerIntegrationTest.java
@@@ -1,421 -1,0 +1,478 @@@
 +/*
 + *
 + *  Licensed to the Apache Software Foundation (ASF) under one
 + *  or more contributor license agreements.  See the NOTICE file
 + *  distributed with this work for additional information
 + *  regarding copyright ownership.  The ASF licenses this file
 + *  to you under the Apache License, Version 2.0 (the
 + *  "License"); you may not use this file except in compliance
 + *  with the License.  You may obtain a copy of the License at
 + *
 + *      http://www.apache.org/licenses/LICENSE-2.0
 + *
 + *  Unless required by applicable law or agreed to in writing, software
 + *  distributed under the License is distributed on an "AS IS" BASIS,
 + *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 + *  See the License for the specific language governing permissions and
 + *  limitations under the License.
 + *
 + */
 +
 +package org.apache.metron.profiler.storm.integration;
 +
 +import org.adrianwalker.multilinestring.Multiline;
- import org.apache.hadoop.hbase.Cell;
- import org.apache.hadoop.hbase.client.Put;
- import org.apache.hadoop.hbase.util.Bytes;
++import org.apache.commons.io.FileUtils;
 +import org.apache.metron.common.Constants;
- import org.apache.metron.common.utils.SerDeUtils;
 +import org.apache.metron.hbase.mock.MockHBaseTableProvider;
 +import org.apache.metron.hbase.mock.MockHTable;
 +import org.apache.metron.integration.BaseIntegrationTest;
 +import org.apache.metron.integration.ComponentRunner;
 +import org.apache.metron.integration.UnableToStartException;
 +import org.apache.metron.integration.components.FluxTopologyComponent;
 +import org.apache.metron.integration.components.KafkaComponent;
 +import org.apache.metron.integration.components.ZKServerComponent;
- import org.apache.metron.profiler.ProfileMeasurement;
- import org.apache.metron.profiler.hbase.ColumnBuilder;
- import org.apache.metron.profiler.hbase.RowKeyBuilder;
- import org.apache.metron.profiler.hbase.SaltyRowKeyBuilder;
- import org.apache.metron.profiler.hbase.ValueOnlyColumnBuilder;
++import org.apache.metron.profiler.client.stellar.FixedLookback;
++import org.apache.metron.profiler.client.stellar.GetProfile;
++import org.apache.metron.profiler.client.stellar.WindowLookback;
++import org.apache.metron.statistics.OnlineStatisticsProvider;
++import org.apache.metron.stellar.common.DefaultStellarStatefulExecutor;
++import org.apache.metron.stellar.common.StellarStatefulExecutor;
++import org.apache.metron.stellar.dsl.Context;
++import org.apache.metron.stellar.dsl.functions.resolver.SimpleFunctionResolver;
 +import org.apache.storm.Config;
 +import org.junit.After;
 +import org.junit.AfterClass;
 +import org.junit.Before;
 +import org.junit.BeforeClass;
 +import org.junit.Test;
++import org.slf4j.Logger;
++import org.slf4j.LoggerFactory;
 +
 +import java.io.File;
 +import java.io.UnsupportedEncodingException;
- import java.util.ArrayList;
++import java.lang.invoke.MethodHandles;
 +import java.util.Arrays;
++import java.util.Collections;
++import java.util.HashMap;
 +import java.util.List;
++import java.util.Map;
 +import java.util.Properties;
 +import java.util.concurrent.TimeUnit;
 +
 +import static com.google.code.tempusfugit.temporal.Duration.seconds;
 +import static com.google.code.tempusfugit.temporal.Timeout.timeout;
 +import static com.google.code.tempusfugit.temporal.WaitFor.waitOrTimeout;
- import static org.junit.Assert.assertArrayEquals;
++import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_COLUMN_FAMILY;
++import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE;
++import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_HBASE_TABLE_PROVIDER;
++import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD;
++import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_PERIOD_UNITS;
++import static org.apache.metron.profiler.client.stellar.ProfilerClientConfig.PROFILER_SALT_DIVISOR;
 +import static org.junit.Assert.assertEquals;
 +import static org.junit.Assert.assertTrue;
 +
 +/**
 + * An integration test of the Profiler topology.
 + */
 +public class ProfilerIntegrationTest extends BaseIntegrationTest {
 +
++  private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 +  private static final String TEST_RESOURCES = "../../metron-analytics/metron-profiler-storm/src/test";
 +  private static final String FLUX_PATH = "src/main/flux/profiler/remote.yaml";
 +
 +  public static final long startAt = 10;
 +  public static final String entity = "10.0.0.1";
 +
 +  private static final String tableName = "profiler";
 +  private static final String columnFamily = "P";
 +  private static final String inputTopic = Constants.INDEXING_TOPIC;
 +  private static final String outputTopic = "profiles";
 +  private static final int saltDivisor = 10;
 +
-   private static final long windowLagMillis = TimeUnit.SECONDS.toMillis(1);
-   private static final long windowDurationMillis = TimeUnit.SECONDS.toMillis(5);
-   private static final long periodDurationMillis = TimeUnit.SECONDS.toMillis(10);
-   private static final long profileTimeToLiveMillis = TimeUnit.SECONDS.toMillis(15);
++  private static final long periodDurationMillis = TimeUnit.SECONDS.toMillis(20);
++  private static final long windowLagMillis = TimeUnit.SECONDS.toMillis(10);
++  private static final long windowDurationMillis = TimeUnit.SECONDS.toMillis(10);
++  private static final long profileTimeToLiveMillis = TimeUnit.SECONDS.toMillis(20);
++
 +  private static final long maxRoutesPerBolt = 100000;
 +
-   private static ColumnBuilder columnBuilder;
 +  private static ZKServerComponent zkComponent;
 +  private static FluxTopologyComponent fluxComponent;
 +  private static KafkaComponent kafkaComponent;
 +  private static ConfigUploadComponent configUploadComponent;
 +  private static ComponentRunner runner;
 +  private static MockHTable profilerTable;
 +
 +  private static String message1;
 +  private static String message2;
 +  private static String message3;
 +
++  private StellarStatefulExecutor executor;
++
 +  /**
 +   * [
 +   *    org.apache.metron.profiler.ProfileMeasurement,
 +   *    org.apache.metron.profiler.ProfilePeriod,
 +   *    org.apache.metron.common.configuration.profiler.ProfileResult,
 +   *    org.apache.metron.common.configuration.profiler.ProfileResultExpressions,
 +   *    org.apache.metron.common.configuration.profiler.ProfileTriageExpressions,
 +   *    org.apache.metron.common.configuration.profiler.ProfilerConfig,
 +   *    org.apache.metron.common.configuration.profiler.ProfileConfig,
 +   *    org.json.simple.JSONObject,
++   *    org.json.simple.JSONArray,
 +   *    java.util.LinkedHashMap,
 +   *    org.apache.metron.statistics.OnlineStatisticsProvider
 +   *  ]
 +   */
 +  @Multiline
 +  private static String kryoSerializers;
 +
-   /**
-    * The Profiler can generate profiles based on processing time.  With processing time,
-    * the Profiler builds profiles based on when the telemetry was processed.
-    *
-    * <p>Not defining a 'timestampField' within the Profiler configuration tells the
Profiler
-    * to use processing time.
-    */
 +  @Test
 +  public void testProcessingTime() throws Exception {
++    uploadConfigToZookeeper(TEST_RESOURCES + "/config/zookeeper/processing-time-test");
 +
-     // upload the config to zookeeper
-     uploadConfig(TEST_RESOURCES + "/config/zookeeper/processing-time-test");
- 
-     // start the topology and write test messages to kafka
++    // start the topology and write 3 test messages to kafka
 +    fluxComponent.submitTopology();
- 
-     // the messages that will be applied to the profile
 +    kafkaComponent.writeMessages(inputTopic, message1);
 +    kafkaComponent.writeMessages(inputTopic, message2);
 +    kafkaComponent.writeMessages(inputTopic, message3);
 +
-     // storm needs at least one message to close its event window
++    // retrieve the profile measurement using PROFILE_GET
++    String profileGetExpression = "PROFILE_GET('processing-time-test', '10.0.0.1', PROFILE_FIXED('5',
'MINUTES'))";
++    List<Integer> measurements = execute(profileGetExpression, List.class);
++
++    // need to keep checking for measurements until the profiler has flushed one out
 +    int attempt = 0;
-     while(profilerTable.getPutLog().size() == 0 && attempt++ < 10) {
++    while(measurements.size() == 0 && attempt++ < 10) {
 +
-       // sleep, at least beyond the current window
-       Thread.sleep(windowDurationMillis + windowLagMillis);
++      // wait for the profiler to flush
++      long sleep = windowDurationMillis;
++      LOG.debug("Waiting {} millis for profiler to flush", sleep);
++      Thread.sleep(sleep);
 +
-       // send another message to help close the current event window
++      // write another message to advance time. this ensures we are testing the 'normal'
flush mechanism.
++      // if we do not send additional messages to advance time, then it is the profile TTL
mechanism which
++      // will ultimately flush the profile
 +      kafkaComponent.writeMessages(inputTopic, message2);
++
++      // try again to retrieve the profile measurement using PROFILE_GET
++      measurements = execute(profileGetExpression, List.class);
 +    }
 +
-     // validate what was flushed
-     List<Integer> actuals = read(
-             profilerTable.getPutLog(),
-             columnFamily,
-             columnBuilder.getColumnQualifier("value"),
-             Integer.class);
-     assertEquals(1, actuals.size());
-     assertTrue(actuals.get(0) >= 3);
++    // expect to see only 1 measurement, but could be more (one for each period) depending
on
++    // how long we waited for the flush to occur
++    assertTrue(measurements.size() > 0);
++
++    // the profile should have counted at least 3 messages; the 3 test messages that were
sent.
++    // the count could be higher due to the test messages we sent to advance time.
++    assertTrue(measurements.get(0) >= 3);
 +  }
 +
-   /**
-    * The Profiler can generate profiles using event time.  With event time processing,
-    * the Profiler uses timestamps contained in the source telemetry.
-    *
-    * <p>Defining a 'timestampField' within the Profiler configuration tells the Profiler
-    * from which field the timestamp should be extracted.
-    */
 +  @Test
-   public void testEventTime() throws Exception {
- 
-     // upload the profiler config to zookeeper
-     uploadConfig(TEST_RESOURCES + "/config/zookeeper/event-time-test");
++  public void testProcessingTimeWithTimeToLiveFlush() throws Exception {
++    uploadConfigToZookeeper(TEST_RESOURCES + "/config/zookeeper/processing-time-test");
 +
-     // start the topology and write test messages to kafka
++    // start the topology and write 3 test messages to kafka
 +    fluxComponent.submitTopology();
 +    kafkaComponent.writeMessages(inputTopic, message1);
 +    kafkaComponent.writeMessages(inputTopic, message2);
 +    kafkaComponent.writeMessages(inputTopic, message3);
 +
-     // wait until the profile is flushed
-     waitOrTimeout(() -> profilerTable.getPutLog().size() > 0, timeout(seconds(90)));
++    // wait a bit beyond the window lag before writing another message.  this allows storm's
window manager to close
++    // the event window, which then lets the profiler processes the previous messages.
++    long sleep = windowLagMillis + periodDurationMillis;
++    LOG.debug("Waiting {} millis before sending message to close window", sleep);
++    Thread.sleep(sleep);
++    kafkaComponent.writeMessages(inputTopic, message3);
++
++    // retrieve the profile measurement using PROFILE_GET
++    String profileGetExpression = "PROFILE_GET('processing-time-test', '10.0.0.1', PROFILE_FIXED('5',
'MINUTES'))";
++    List<Integer> measurements = execute(profileGetExpression, List.class);
 +
-     List<Put> puts = profilerTable.getPutLog();
-     assertEquals(1, puts.size());
++    // need to keep checking for measurements until the profiler has flushed one out
++    int attempt = 0;
++    while(measurements.size() == 0 && attempt++ < 10) {
++
++      // wait for the profiler to flush
++      sleep = windowDurationMillis;
++      LOG.debug("Waiting {} millis for profiler to flush", sleep);
++      Thread.sleep(sleep);
++
++      // do not write additional messages to advance time. this ensures that we are testing
the "time to live"
++      // flush mechanism. the TTL setting defines when the profile will be flushed
++
++      // try again to retrieve the profile measurement
++      measurements = execute(profileGetExpression, List.class);
++    }
++
++    // expect to see only 1 measurement, but could be more (one for each period) depending
on
++    // how long we waited for the flush to occur
++    assertTrue(measurements.size() > 0);
 +
-     // inspect the row key to ensure the profiler used event time correctly.  the timestamp
-     // embedded in the row key should match those in the source telemetry
-     byte[] expectedRowKey = generateExpectedRowKey("event-time-test", entity, startAt);
-     byte[] actualRowKey = puts.get(0).getRow();
-     assertArrayEquals(failMessage(expectedRowKey, actualRowKey), expectedRowKey, actualRowKey);
++    // the profile should have counted 3 messages; the 3 test messages that were sent
++    assertEquals(3, measurements.get(0).intValue());
++  }
++
++  @Test
++  public void testEventTime() throws Exception {
++    uploadConfigToZookeeper(TEST_RESOURCES + "/config/zookeeper/event-time-test");
++
++    // start the topology and write test messages to kafka
++    fluxComponent.submitTopology();
++    List<String> messages = FileUtils.readLines(new File("src/test/resources/telemetry.json"));
++    kafkaComponent.writeMessages(inputTopic, messages);
++
++    long timestamp = System.currentTimeMillis();
++    LOG.debug("Attempting to close window period by sending message with timestamp = {}",
timestamp);
++    kafkaComponent.writeMessages(inputTopic, getMessage("192.168.66.1", timestamp));
++    kafkaComponent.writeMessages(inputTopic, getMessage("192.168.138.158", timestamp));
++
++    // create the 'window' that looks up to 5 hours before the max timestamp contained in
the test data
++    assign("maxTimestamp", "1530978728982L");
++    assign("window", "PROFILE_WINDOW('from 5 hours ago', maxTimestamp)");
++
++    // wait until the profile flushes both periods.  the first period will flush immediately
as subsequent messages
++    // advance time.  the next period contains all of the remaining messages, so there are
no other messages to
++    // advance time.  because of this the next period only flushes after the time-to-live
expires
++    waitOrTimeout(() -> profilerTable.getPutLog().size() >= 6, timeout(seconds(90)));
++    {
++      // there are 14 messages in the first period and 12 in the next where ip_src_addr
= 192.168.66.1
++      List results = execute("PROFILE_GET('count-by-ip', '192.168.66.1', window)", List.class);
++      assertEquals(14, results.get(0));
++      assertEquals(12, results.get(1));
++    }
++    {
++      // there are 36 messages in the first period and 38 in the next where ip_src_addr
= 192.168.138.158
++      List results = execute("PROFILE_GET('count-by-ip', '192.168.138.158', window)", List.class);
++      assertEquals(36, results.get(0));
++      assertEquals(38, results.get(1));
++    }
++    {
++      // in all there are 50 messages in the first period and 50 messages in the next
++      List results = execute("PROFILE_GET('total-count', 'total', window)", List.class);
++      assertEquals(50, results.get(0));
++      assertEquals(50, results.get(1));
++    }
 +  }
 +
 +  /**
 +   * The result produced by a Profile has to be serializable within Storm. If the result
is not
 +   * serializable the topology will crash and burn.
 +   *
 +   * This test ensures that if a profile returns a STATS object created using the STATS_INIT
and
 +   * STATS_ADD functions, that it can be correctly serialized and persisted.
 +   */
 +  @Test
 +  public void testProfileWithStatsObject() throws Exception {
- 
-     // upload the profiler config to zookeeper
-     uploadConfig(TEST_RESOURCES + "/config/zookeeper/profile-with-stats");
++    uploadConfigToZookeeper(TEST_RESOURCES + "/config/zookeeper/profile-with-stats");
 +
 +    // start the topology and write test messages to kafka
 +    fluxComponent.submitTopology();
-     kafkaComponent.writeMessages(inputTopic, message1);
-     kafkaComponent.writeMessages(inputTopic, message2);
-     kafkaComponent.writeMessages(inputTopic, message3);
++    List<String> messages = FileUtils.readLines(new File("src/test/resources/telemetry.json"));
++    kafkaComponent.writeMessages(inputTopic, messages);
 +
 +    // wait until the profile is flushed
 +    waitOrTimeout(() -> profilerTable.getPutLog().size() > 0, timeout(seconds(90)));
 +
-     // ensure that a value was persisted in HBase
-     List<Put> puts = profilerTable.getPutLog();
-     assertEquals(1, puts.size());
- 
-     // generate the expected row key. only the profile name, entity, and period are used
to generate the row key
-     ProfileMeasurement measurement = new ProfileMeasurement()
-             .withProfileName("profile-with-stats")
-             .withEntity("global")
-             .withPeriod(startAt, periodDurationMillis, TimeUnit.MILLISECONDS);
-     RowKeyBuilder rowKeyBuilder = new SaltyRowKeyBuilder(saltDivisor, periodDurationMillis,
TimeUnit.MILLISECONDS);
-     byte[] expectedRowKey = rowKeyBuilder.rowKey(measurement);
- 
-     // ensure the correct row key was generated
-     byte[] actualRowKey = puts.get(0).getRow();
-     assertArrayEquals(failMessage(expectedRowKey, actualRowKey), expectedRowKey, actualRowKey);
++    // validate the measurements written by the batch profiler using `PROFILE_GET`
++    // the 'window' looks up to 5 hours before the max timestamp contained in the test data
++    assign("maxTimestamp", "1530978728982L");
++    assign("window", "PROFILE_WINDOW('from 5 hours ago', maxTimestamp)");
++
++    // retrieve the stats stored by the profiler
++    List results = execute("PROFILE_GET('profile-with-stats', 'global', window)", List.class);
++    assertTrue(results.size() > 0);
++    assertTrue(results.get(0) instanceof OnlineStatisticsProvider);
 +  }
 +
 +  /**
 +   * Generates an error message for if the byte comparison fails.
 +   *
 +   * @param expected The expected value.
 +   * @param actual The actual value.
 +   * @return
 +   * @throws UnsupportedEncodingException
 +   */
 +  private String failMessage(byte[] expected, byte[] actual) throws UnsupportedEncodingException
{
 +    return String.format("expected '%s', got '%s'",
 +              new String(expected, "UTF-8"),
 +              new String(actual, "UTF-8"));
 +  }
 +
-   /**
-    * Generates the expected row key.
-    *
-    * @param profileName The name of the profile.
-    * @param entity The entity.
-    * @param whenMillis A timestamp in epoch milliseconds.
-    * @return A row key.
-    */
-   private byte[] generateExpectedRowKey(String profileName, String entity, long whenMillis)
{
- 
-     // only the profile name, entity, and period are used to generate the row key
-     ProfileMeasurement measurement = new ProfileMeasurement()
-             .withProfileName(profileName)
-             .withEntity(entity)
-             .withPeriod(whenMillis, periodDurationMillis, TimeUnit.MILLISECONDS);
- 
-     // build the row key
-     RowKeyBuilder rowKeyBuilder = new SaltyRowKeyBuilder(saltDivisor, periodDurationMillis,
TimeUnit.MILLISECONDS);
-     return rowKeyBuilder.rowKey(measurement);
-   }
- 
-   /**
-    * Reads a value written by the Profiler.
-    *
-    * @param family The column family.
-    * @param qualifier The column qualifier.
-    * @param clazz The expected type of the value.
-    * @param <T> The expected type of the value.
-    * @return The value written by the Profiler.
-    */
-   private <T> List<T> read(List<Put> puts, String family, byte[] qualifier,
Class<T> clazz) {
-     List<T> results = new ArrayList<>();
- 
-     for(Put put: puts) {
-       List<Cell> cells = put.get(Bytes.toBytes(family), qualifier);
-       for(Cell cell : cells) {
-         T value = SerDeUtils.fromBytes(cell.getValue(), clazz);
-         results.add(value);
-       }
-     }
- 
-     return results;
++  private static String getMessage(String ipSource, long timestamp) {
++    return new MessageBuilder()
++            .withField("ip_src_addr", ipSource)
++            .withField("timestamp", timestamp)
++            .build()
++            .toJSONString();
 +  }
 +
 +  @BeforeClass
 +  public static void setupBeforeClass() throws UnableToStartException {
 +
 +    // create some messages that contain a timestamp - a really old timestamp; close to
1970
-     message1 = new MessageBuilder()
-             .withField("ip_src_addr", entity)
-             .withField("timestamp", startAt)
-             .build()
-             .toJSONString();
- 
-     message2 = new MessageBuilder()
-             .withField("ip_src_addr", entity)
-             .withField("timestamp", startAt + 100)
-             .build()
-             .toJSONString();
- 
-     message3 = new MessageBuilder()
-             .withField("ip_src_addr", entity)
-             .withField("timestamp", startAt + (windowDurationMillis * 2))
-             .build()
-             .toJSONString();
- 
-     columnBuilder = new ValueOnlyColumnBuilder(columnFamily);
++    message1 = getMessage(entity, startAt);
++    message2 = getMessage(entity, startAt + 100);
++    message3 = getMessage(entity, startAt + (windowDurationMillis * 2));
 +
 +    // storm topology properties
 +    final Properties topologyProperties = new Properties() {{
 +
 +      // storm settings
 +      setProperty("profiler.workers", "1");
 +      setProperty("profiler.executors", "0");
++
 +      setProperty(Config.TOPOLOGY_AUTO_CREDENTIALS, "[]");
 +      setProperty(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, "60");
 +      setProperty(Config.TOPOLOGY_MAX_SPOUT_PENDING, "100000");
 +
 +      // ensure tuples are serialized during the test, otherwise serialization problems
 +      // will not be found until the topology is run on a cluster with multiple workers
 +      setProperty(Config.TOPOLOGY_TESTING_ALWAYS_TRY_SERIALIZE, "true");
 +      setProperty(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, "false");
 +      setProperty(Config.TOPOLOGY_KRYO_REGISTER, kryoSerializers);
 +
 +      // kafka settings
 +      setProperty("profiler.input.topic", inputTopic);
 +      setProperty("profiler.output.topic", outputTopic);
-       setProperty("kafka.start", "UNCOMMITTED_EARLIEST");
++      setProperty("kafka.start", "EARLIEST");
 +      setProperty("kafka.security.protocol", "PLAINTEXT");
 +
 +      // hbase settings
 +      setProperty("profiler.hbase.salt.divisor", Integer.toString(saltDivisor));
 +      setProperty("profiler.hbase.table", tableName);
 +      setProperty("profiler.hbase.column.family", columnFamily);
 +      setProperty("profiler.hbase.batch", "10");
 +      setProperty("profiler.hbase.flush.interval.seconds", "1");
 +      setProperty("hbase.provider.impl", "" + MockHBaseTableProvider.class.getName());
 +
 +      // profile settings
 +      setProperty("profiler.period.duration", Long.toString(periodDurationMillis));
 +      setProperty("profiler.period.duration.units", "MILLISECONDS");
 +      setProperty("profiler.ttl", Long.toString(profileTimeToLiveMillis));
 +      setProperty("profiler.ttl.units", "MILLISECONDS");
 +      setProperty("profiler.window.duration", Long.toString(windowDurationMillis));
 +      setProperty("profiler.window.duration.units", "MILLISECONDS");
 +      setProperty("profiler.window.lag", Long.toString(windowLagMillis));
 +      setProperty("profiler.window.lag.units", "MILLISECONDS");
 +      setProperty("profiler.max.routes.per.bolt", Long.toString(maxRoutesPerBolt));
 +    }};
 +
 +    // create the mock table
 +    profilerTable = (MockHTable) MockHBaseTableProvider.addToCache(tableName, columnFamily);
 +
 +    zkComponent = getZKServerComponent(topologyProperties);
 +
 +    // create the input and output topics
 +    kafkaComponent = getKafkaComponent(topologyProperties, Arrays.asList(
 +            new KafkaComponent.Topic(inputTopic, 1),
 +            new KafkaComponent.Topic(outputTopic, 1)));
 +
 +    // upload profiler configuration to zookeeper
 +    configUploadComponent = new ConfigUploadComponent()
 +            .withTopologyProperties(topologyProperties);
 +
 +    // load flux definition for the profiler topology
 +    fluxComponent = new FluxTopologyComponent.Builder()
 +            .withTopologyLocation(new File(FLUX_PATH))
 +            .withTopologyName("profiler")
 +            .withTopologyProperties(topologyProperties)
 +            .build();
 +
 +    // start all components
 +    runner = new ComponentRunner.Builder()
 +            .withComponent("zk",zkComponent)
 +            .withComponent("kafka", kafkaComponent)
 +            .withComponent("config", configUploadComponent)
 +            .withComponent("storm", fluxComponent)
 +            .withMillisecondsBetweenAttempts(15000)
 +            .withNumRetries(10)
 +            .withCustomShutdownOrder(new String[] {"storm","config","kafka","zk"})
 +            .build();
 +    runner.start();
 +  }
 +
 +  @AfterClass
 +  public static void tearDownAfterClass() throws Exception {
 +    MockHBaseTableProvider.clear();
 +    if (runner != null) {
 +      runner.stop();
 +    }
 +  }
 +
 +  @Before
 +  public void setup() {
 +    // create the mock table
 +    profilerTable = (MockHTable) MockHBaseTableProvider.addToCache(tableName, columnFamily);
++
++    // global properties
++    Map<String, Object> global = new HashMap<String, Object>() {{
++      put(PROFILER_HBASE_TABLE.getKey(), tableName);
++      put(PROFILER_COLUMN_FAMILY.getKey(), columnFamily);
++      put(PROFILER_HBASE_TABLE_PROVIDER.getKey(), MockHBaseTableProvider.class.getName());
++
++      // client needs to use the same period duration
++      put(PROFILER_PERIOD.getKey(), Long.toString(periodDurationMillis));
++      put(PROFILER_PERIOD_UNITS.getKey(), "MILLISECONDS");
++
++      // client needs to use the same salt divisor
++      put(PROFILER_SALT_DIVISOR.getKey(), saltDivisor);
++    }};
++
++    // create the stellar execution environment
++    executor = new DefaultStellarStatefulExecutor(
++            new SimpleFunctionResolver()
++                    .withClass(GetProfile.class)
++                    .withClass(FixedLookback.class)
++                    .withClass(WindowLookback.class),
++            new Context.Builder()
++                    .with(Context.Capabilities.GLOBAL_CONFIG, () -> global)
++                    .build());
 +  }
 +
 +  @After
 +  public void tearDown() throws Exception {
 +    MockHBaseTableProvider.clear();
 +    profilerTable.clear();
 +    if (runner != null) {
 +      runner.reset();
 +    }
 +  }
 +
 +  /**
 +   * Uploads config values to Zookeeper.
 +   * @param path The path on the local filesystem to the config values.
 +   * @throws Exception
 +   */
-   public void uploadConfig(String path) throws Exception {
++  public void uploadConfigToZookeeper(String path) throws Exception {
 +    configUploadComponent
 +            .withGlobalConfiguration(path)
 +            .withProfilerConfiguration(path)
 +            .update();
 +  }
++
++  /**
++   * Assign a value to the result of an expression.
++   *
++   * @param var The variable to assign.
++   * @param expression The expression to execute.
++   */
++  private void assign(String var, String expression) {
++    executor.assign(var, expression, Collections.emptyMap());
++  }
++
++  /**
++   * Execute a Stellar expression.
++   *
++   * @param expression The Stellar expression to execute.
++   * @param clazz
++   * @param <T>
++   * @return The result of executing the Stellar expression.
++   */
++  private <T> T execute(String expression, Class<T> clazz) {
++    T results = executor.execute(expression, Collections.emptyMap(), clazz);
++
++    LOG.debug("{} = {}", expression, results);
++    return results;
++  }
 +}

http://git-wip-us.apache.org/repos/asf/metron/blob/cad2f408/metron-analytics/metron-profiler-storm/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --cc metron-analytics/metron-profiler-storm/src/test/resources/log4j.properties
index 541f368,0000000..1c2359a
mode 100644,000000..100644
--- a/metron-analytics/metron-profiler-storm/src/test/resources/log4j.properties
+++ b/metron-analytics/metron-profiler-storm/src/test/resources/log4j.properties
@@@ -1,34 -1,0 +1,32 @@@
 +#
 +#
 +#  Licensed to the Apache Software Foundation (ASF) under one
 +#  or more contributor license agreements.  See the NOTICE file
 +#  distributed with this work for additional information
 +#  regarding copyright ownership.  The ASF licenses this file
 +#  to you under the Apache License, Version 2.0 (the
 +#  "License"); you may not use this file except in compliance
 +#  with the License.  You may obtain a copy of the License at
 +#
 +#      http://www.apache.org/licenses/LICENSE-2.0
 +#
 +#  Unless required by applicable law or agreed to in writing, software
 +#  distributed under the License is distributed on an "AS IS" BASIS,
 +#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 +#  See the License for the specific language governing permissions and
 +#  limitations under the License.
 +#
 +#
 +
 +# Root logger option
 +log4j.rootLogger=ERROR, stdout
 +
 +# Direct log messages to stdout
 +log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 +log4j.appender.stdout.Target=System.out
 +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 +log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n
- log4j.appender.stdout.filter.1=org.apache.log4j.varia.StringMatchFilter
- log4j.appender.stdout.filter.1.StringToMatch=Connection timed out
- log4j.appender.stdout.filter.1.AcceptOnMatch=false
- log4j.appender.stdout.filter.2=org.apache.log4j.varia.StringMatchFilter
- log4j.appender.stdout.filter.2.StringToMatch=Background
- log4j.appender.stdout.filter.2.AcceptOnMatch=false
++
++# uncomment below to help debug tests
++#log4j.logger.org.apache.metron.profiler=ALL
++#log4j.logger.org.apache.storm.windowing=ALL


Mime
View raw message