METRON-1741 Move REPL Port of Profiler to Separate Project (nickwallen) closes apache/metron#1170
Project: http://git-wip-us.apache.org/repos/asf/metron/repo Commit: http://git-wip-us.apache.org/repos/asf/metron/commit/9455c4ee Tree: http://git-wip-us.apache.org/repos/asf/metron/tree/9455c4ee Diff: http://git-wip-us.apache.org/repos/asf/metron/diff/9455c4ee Branch: refs/heads/master Commit: 9455c4ee3f6962c47140522dc4c0f2a243695e8d Parents: f5eaef3 Author: nickwallen <n...@nickallen.org> Authored: Tue Sep 4 17:16:29 2018 -0400 Committer: nickallen <nickal...@apache.org> Committed: Tue Sep 4 17:16:29 2018 -0400 ---------------------------------------------------------------------- .../client/stellar/ProfilerFunctions.java | 283 ---------------- .../client/stellar/ProfilerFunctionsTest.java | 334 ------------------- .../metron/profiler/StandAloneProfiler.java | 174 ---------- .../metron/profiler/StandAloneProfilerTest.java | 255 -------------- metron-analytics/metron-profiler-repl/README.md | 178 ++++++++++ metron-analytics/metron-profiler-repl/pom.xml | 150 +++++++++ .../src/main/assembly/assembly.xml | 36 ++ .../metron/profiler/repl/ProfilerFunctions.java | 278 +++++++++++++++ .../profiler/repl/StandAloneProfiler.java | 176 ++++++++++ .../profiler/repl/ProfilerFunctionsTest.java | 332 ++++++++++++++++++ .../profiler/repl/StandAloneProfilerTest.java | 255 ++++++++++++++ metron-analytics/pom.xml | 1 + .../common-services/METRON/CURRENT/metainfo.xml | 3 + .../packaging/docker/deb-docker/pom.xml | 6 + .../docker/rpm-docker/SPECS/metron.spec | 21 ++ .../packaging/docker/rpm-docker/pom.xml | 6 + .../metron-common/src/main/scripts/stellar | 3 +- 17 files changed, 1444 insertions(+), 1047 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/ProfilerFunctions.java ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/ProfilerFunctions.java b/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/ProfilerFunctions.java deleted file mode 100644 index d6afe1d..0000000 --- a/metron-analytics/metron-profiler-client/src/main/java/org/apache/metron/profiler/client/stellar/ProfilerFunctions.java +++ /dev/null @@ -1,283 +0,0 @@ -/* - * - * 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.client.stellar; - -import org.apache.commons.lang3.ClassUtils; -import org.apache.metron.common.configuration.profiler.ProfilerConfig; -import org.apache.metron.common.utils.JSONUtils; -import org.apache.metron.profiler.ProfileMeasurement; -import org.apache.metron.profiler.StandAloneProfiler; -import org.apache.metron.stellar.dsl.Context; -import org.apache.metron.stellar.dsl.ParseException; -import org.apache.metron.stellar.dsl.Stellar; -import org.apache.metron.stellar.dsl.StellarFunction; -import org.json.simple.JSONArray; -import org.json.simple.JSONObject; -import org.json.simple.parser.JSONParser; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.lang.invoke.MethodHandles; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; - -import static java.lang.String.format; -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.Util.getArg; -import static org.apache.metron.stellar.dsl.Context.Capabilities.GLOBAL_CONFIG; - -/** - * Stellar functions that allow interaction with the core Profiler components - * through the Stellar REPL. - */ -public class ProfilerFunctions { - - private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); - - @Stellar( - namespace="PROFILER", - name="INIT", - description="Creates a local profile runner that can execute profiles.", - params={ - "config", "The profiler configuration as a string." - }, - returns="A local profile runner." - ) - public static class ProfilerInit implements StellarFunction { - - @Override - public void initialize(Context context) { - } - - @Override - public boolean isInitialized() { - return true; - } - - @Override - public Object apply(List<Object> args, Context context) { - @SuppressWarnings("unchecked") - Map<String, Object> global = (Map<String, Object>) context.getCapability(GLOBAL_CONFIG, false) - .orElse(Collections.emptyMap()); - - // how long is the profile period? - long duration = PROFILER_PERIOD.getOrDefault(global, PROFILER_PERIOD.getDefault(), Long.class); - String configuredUnits = PROFILER_PERIOD_UNITS.getOrDefault(global, PROFILER_PERIOD_UNITS.getDefault(), String.class); - long periodDurationMillis = TimeUnit.valueOf(configuredUnits).toMillis(duration); - - // user must provide the configuration for the profiler - String arg0 = getArg(0, String.class, args); - ProfilerConfig profilerConfig; - try { - profilerConfig = JSONUtils.INSTANCE.load(arg0, ProfilerConfig.class); - - } catch(IOException e) { - throw new IllegalArgumentException("Invalid profiler configuration", e); - } - - // the TTL and max routes do not matter here - long profileTimeToLiveMillis = Long.MAX_VALUE; - long maxNumberOfRoutes = Long.MAX_VALUE; - return new StandAloneProfiler(profilerConfig, periodDurationMillis, profileTimeToLiveMillis, maxNumberOfRoutes, context); - } - } - - @Stellar( - namespace="PROFILER", - name="APPLY", - description="Apply a message to a local profile runner.", - params={ - "message(s)", "The message to apply; a JSON string or list of JSON strings.", - "profiler", "A local profile runner returned by PROFILER_INIT." - }, - returns="The local profile runner." - ) - public static class ProfilerApply implements StellarFunction { - - private JSONParser parser; - - @Override - public void initialize(Context context) { - parser = new JSONParser(); - } - - @Override - public boolean isInitialized() { - return parser != null; - } - - @Override - public Object apply(List<Object> args, Context context) throws ParseException { - - // the use can pass in one or more messages in a few different forms - Object arg0 = Util.getArg(0, Object.class, args); - List<JSONObject> messages = getMessages(arg0); - - // user must provide the stand alone profiler - StandAloneProfiler profiler = Util.getArg(1, StandAloneProfiler.class, args); - for (JSONObject message : messages) { - profiler.apply(message); - } - - return profiler; - } - - /** - * Gets a message or messages from the function arguments. - * - * @param arg The function argument containing the message(s). - * @return A list of messages - */ - private List<JSONObject> getMessages(Object arg) { - List<JSONObject> messages; - - if (arg instanceof String) { - messages = getMessagesFromString((String) arg); - - } else if (arg instanceof Iterable) { - messages = getMessagesFromIterable((Iterable<String>) arg); - - } else if (arg instanceof JSONObject) { - messages = Collections.singletonList((JSONObject) arg); - - } else { - throw new IllegalArgumentException(format("invalid message: found '%s', expected String, List, or JSONObject", - ClassUtils.getShortClassName(arg, "null"))); - } - - return messages; - } - - /** - * Gets a message or messages from a List - * - * @param strings The function argument that is a bunch of strings. - * @return A list of messages. - */ - private List<JSONObject> getMessagesFromIterable(Iterable<String> strings) { - List<JSONObject> messages = new ArrayList<>(); - - // the user pass in a list of strings - for (String str : strings) { - messages.addAll(getMessagesFromString(str)); - } - - return messages; - } - - /** - * Gets a message or messages from a String argument. - * - * @param arg0 The function argument is just a List. - * @return A list of messages. - */ - private List<JSONObject> getMessagesFromString(String arg0) { - List<JSONObject> messages = new ArrayList<>(); - - try { - Object parsedArg0 = parser.parse(arg0); - if (parsedArg0 instanceof JSONObject) { - // if the string only contains one message - messages.add((JSONObject) parsedArg0); - - } else if (parsedArg0 instanceof JSONArray) { - // if the string contains multiple messages - JSONArray jsonArray = (JSONArray) parsedArg0; - for (Object item : jsonArray) { - messages.addAll(getMessages(item)); - } - - } else { - throw new IllegalArgumentException(format("invalid message: found '%s', expected JSONObject or JSONArray", - ClassUtils.getShortClassName(parsedArg0, "null"))); - } - - } catch (org.json.simple.parser.ParseException e) { - throw new IllegalArgumentException(format("invalid message: '%s'", e.getMessage()), e); - } - - return messages; - } - } - - @Stellar( - namespace="PROFILER", - name="FLUSH", - description="Flush a local profile runner.", - params={ - "profiler", "A local profile runner returned by PROFILER_INIT." - }, - returns="A list of the profile values." - ) - public static class ProfilerFlush implements StellarFunction { - - @Override - public void initialize(Context context) { - } - - @Override - public boolean isInitialized() { - return true; - } - - @Override - public Object apply(List<Object> args, Context context) throws ParseException { - - // user must provide the stand-alone profiler - StandAloneProfiler profiler = Util.getArg(0, StandAloneProfiler.class, args); - if(profiler == null) { - throw new IllegalArgumentException(format("expected the profiler returned by PROFILER_INIT, found null")); - } - - // transform the profile measurements into maps to simplify manipulation in stellar - List<Map<String, Object>> measurements = new ArrayList<>(); - for(ProfileMeasurement m : profiler.flush()) { - - // create a map for the profile period - Map<String, Object> period = new HashMap<>(); - period.put("period", m.getPeriod().getPeriod()); - period.put("start", m.getPeriod().getStartTimeMillis()); - period.put("duration", m.getPeriod().getDurationMillis()); - period.put("end", m.getPeriod().getEndTimeMillis()); - - // create a map for the measurement - Map<String, Object> measurement = new HashMap<>(); - measurement.put("profile", m.getProfileName()); - measurement.put("entity", m.getEntity()); - measurement.put("value", m.getProfileValue()); - measurement.put("groups", m.getGroups()); - measurement.put("period", period); - - measurements.add(measurement); - } - - return measurements; - } - } -} http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/ProfilerFunctionsTest.java ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/ProfilerFunctionsTest.java b/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/ProfilerFunctionsTest.java deleted file mode 100644 index 1670e8c..0000000 --- a/metron-analytics/metron-profiler-client/src/test/java/org/apache/metron/profiler/client/stellar/ProfilerFunctionsTest.java +++ /dev/null @@ -1,334 +0,0 @@ -/* - * - * 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.client.stellar; - -import org.adrianwalker.multilinestring.Multiline; -import org.apache.metron.profiler.StandAloneProfiler; -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.ParseException; -import org.apache.metron.stellar.dsl.functions.resolver.SimpleFunctionResolver; -import org.json.simple.JSONObject; -import org.json.simple.parser.JSONParser; -import org.junit.Before; -import org.junit.Test; - -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -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.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertSame; - -/** - * Tests the ProfilerFunctions class. - */ -public class ProfilerFunctionsTest { - - /** - * { - * "ip_src_addr": "10.0.0.1", - * "ip_dst_addr": "10.0.0.2", - * "source.type": "test", - * } - */ - @Multiline - private String message; - - /** - * [ - * { - * "ip_src_addr": "10.0.0.1", - * "ip_dst_addr": "10.0.0.2", - * "source.type": "test", - * }, - * { - * "ip_src_addr": "10.0.0.1", - * "ip_dst_addr": "10.0.0.2", - * "source.type": "test", - * }, - * { - * "ip_src_addr": "10.0.0.1", - * "ip_dst_addr": "10.0.0.2", - * "source.type": "test", - * } - * ] - */ - @Multiline - private String messages; - - /** - * { - * "profiles": [ - * { - * "profile": "hello-world", - * "foreach": "ip_src_addr", - * "init": { "count": 0 }, - * "update": { "count": "count + 1" }, - * "result": "count" - * } - * ] - * } - */ - @Multiline - private String helloWorldProfilerDef; - - private static final long periodDuration = 15; - private static final String periodUnits = "MINUTES"; - private StellarStatefulExecutor executor; - private Map<String, Object> state; - - private <T> T run(String expression, Class<T> clazz) { - return executor.execute(expression, state, clazz); - } - - @Before - public void setup() { - state = new HashMap<>(); - - // global properties - Map<String, Object> global = new HashMap<String, Object>() {{ - put(PROFILER_PERIOD.getKey(), Long.toString(periodDuration)); - put(PROFILER_PERIOD_UNITS.getKey(), periodUnits.toString()); - }}; - - // create the stellar execution environment - executor = new DefaultStellarStatefulExecutor( - new SimpleFunctionResolver() - .withClass(ProfilerFunctions.ProfilerInit.class) - .withClass(ProfilerFunctions.ProfilerApply.class) - .withClass(ProfilerFunctions.ProfilerFlush.class), - new Context.Builder() - .with(Context.Capabilities.GLOBAL_CONFIG, () -> global) - .build()); - } - - @Test - public void testProfilerInitNoProfiles() { - state.put("config", "{ \"profiles\" : [] }"); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - assertNotNull(profiler); - assertEquals(0, profiler.getProfileCount()); - assertEquals(0, profiler.getMessageCount()); - assertEquals(0, profiler.getRouteCount()); - } - - @Test - public void testProfilerInitWithProfiles() { - state.put("config", helloWorldProfilerDef); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - assertNotNull(profiler); - assertEquals(1, profiler.getProfileCount()); - assertEquals(0, profiler.getMessageCount()); - assertEquals(0, profiler.getRouteCount()); - } - - @Test(expected = ParseException.class) - public void testProfilerInitNoArgs() { - run("PROFILER_INIT()", StandAloneProfiler.class); - } - - @Test(expected = ParseException.class) - public void testProfilerInitInvalidArg() { - run("PROFILER_INIT({ \"invalid\": 2 })", StandAloneProfiler.class); - } - - @Test - public void testProfilerInitWithNoGlobalConfig() { - state.put("config", helloWorldProfilerDef); - String expression = "PROFILER_INIT(config)"; - - // use an executor with no GLOBAL_CONFIG defined in the context - StellarStatefulExecutor executor = new DefaultStellarStatefulExecutor( - new SimpleFunctionResolver() - .withClass(ProfilerFunctions.ProfilerInit.class) - .withClass(ProfilerFunctions.ProfilerApply.class) - .withClass(ProfilerFunctions.ProfilerFlush.class), - Context.EMPTY_CONTEXT()); - StandAloneProfiler profiler = executor.execute(expression, state, StandAloneProfiler.class); - - assertNotNull(profiler); - assertEquals(1, profiler.getProfileCount()); - assertEquals(0, profiler.getMessageCount()); - assertEquals(0, profiler.getRouteCount()); - } - - @Test - public void testProfilerApplyWithString() { - - // initialize the profiler - state.put("config", helloWorldProfilerDef); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - state.put("profiler", profiler); - - // apply a message to the profiler - state.put("message", message); - StandAloneProfiler result = run("PROFILER_APPLY(message, profiler)", StandAloneProfiler.class); - - // validate - assertSame(profiler, result); - assertEquals(1, profiler.getProfileCount()); - assertEquals(1, profiler.getMessageCount()); - assertEquals(1, profiler.getRouteCount()); - } - - @Test - public void testProfilerApplyWithJSONObject() throws Exception { - - // initialize the profiler - state.put("config", helloWorldProfilerDef); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - state.put("profiler", profiler); - - // apply a message to the profiler - JSONParser parser = new JSONParser(); - JSONObject jsonObject = (JSONObject) parser.parse(message); - state.put("jsonObj", jsonObject); - StandAloneProfiler result = run("PROFILER_APPLY(jsonObj, profiler)", StandAloneProfiler.class); - - // validate - assertSame(profiler, result); - assertEquals(1, profiler.getProfileCount()); - assertEquals(1, profiler.getMessageCount()); - assertEquals(1, profiler.getRouteCount()); - } - - @Test - public void testProfilerApplyWithMultipleMessagesInJSONString() { - - // initialize the profiler - state.put("config", helloWorldProfilerDef); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - state.put("profiler", profiler); - - // apply a message to the profiler - state.put("messages", messages); - StandAloneProfiler result = run("PROFILER_APPLY(messages, profiler)", StandAloneProfiler.class); - - // validate - assertSame(profiler, result); - assertEquals(1, profiler.getProfileCount()); - assertEquals(3, profiler.getMessageCount()); - assertEquals(3, profiler.getRouteCount()); - } - - @Test - public void testProfilerApplyWithListOfMessages() { - - // initialize the profiler - state.put("config", helloWorldProfilerDef); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - state.put("profiler", profiler); - - // apply a message to the profiler - state.put("msg", message); - StandAloneProfiler result = run("PROFILER_APPLY([msg, msg, msg], profiler)", StandAloneProfiler.class); - - // validate - assertSame(profiler, result); - assertEquals(1, profiler.getProfileCount()); - assertEquals(3, profiler.getMessageCount()); - assertEquals(3, profiler.getRouteCount()); - } - - - @Test - public void testProfilerApplyWithEmptyList() { - - // initialize the profiler - state.put("config", helloWorldProfilerDef); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - state.put("profiler", profiler); - - // apply a message to the profiler - state.put("messages", "[ ]"); - StandAloneProfiler result = run("PROFILER_APPLY(messages, profiler)", StandAloneProfiler.class); - - // validate - assertSame(profiler, result); - assertEquals(1, profiler.getProfileCount()); - assertEquals(0, profiler.getMessageCount()); - assertEquals(0, profiler.getRouteCount()); - } - - @Test(expected = ParseException.class) - public void testProfilerApplyWithNoArgs() { - run("PROFILER_APPLY()", StandAloneProfiler.class); - } - - @Test(expected = ParseException.class) - public void testProfilerApplyWithInvalidArg() { - run("PROFILER_APPLY(undefined)", StandAloneProfiler.class); - } - - @Test(expected = ParseException.class) - public void testProfilerApplyWithNullMessage() { - - // initialize the profiler - state.put("config", helloWorldProfilerDef); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - state.put("profiler", profiler); - - // there is no 'messages' variable - should throw exception - run("PROFILER_APPLY(messages, profiler)", StandAloneProfiler.class); - } - - @Test - public void testProfilerFlush() { - - // initialize the profiler - state.put("config", helloWorldProfilerDef); - StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); - state.put("profiler", profiler); - - // apply a message to the profiler - state.put("message", message); - run("PROFILER_APPLY(message, profiler)", StandAloneProfiler.class); - - // flush the profiles - List<Map<String, Object>> measurements = run("PROFILER_FLUSH(profiler)", List.class); - - // validate - assertNotNull(measurements); - assertEquals(1, measurements.size()); - - Map<String, Object> measurement = measurements.get(0); - assertEquals("hello-world", measurement.get("profile")); - assertEquals("10.0.0.1", measurement.get("entity")); - assertEquals(1, measurement.get("value")); - assertEquals(Collections.emptyList(), measurement.get("groups")); - } - - @Test(expected = ParseException.class) - public void testProfilerFlushNoArgs() { - run("PROFILER_FLUSH()", StandAloneProfiler.class); - } - - @Test(expected = ParseException.class) - public void testProfilerFlushInvalidArg() { - run("PROFILER_FLUSH(undefined)", StandAloneProfiler.class); - } -} http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/StandAloneProfiler.java ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/StandAloneProfiler.java b/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/StandAloneProfiler.java deleted file mode 100644 index befa296..0000000 --- a/metron-analytics/metron-profiler-common/src/main/java/org/apache/metron/profiler/StandAloneProfiler.java +++ /dev/null @@ -1,174 +0,0 @@ -/* - * - * 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; - -import org.apache.metron.common.configuration.profiler.ProfilerConfig; -import org.apache.metron.profiler.clock.Clock; -import org.apache.metron.profiler.clock.ClockFactory; -import org.apache.metron.profiler.clock.DefaultClockFactory; -import org.apache.metron.stellar.dsl.Context; -import org.json.simple.JSONObject; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.lang.invoke.MethodHandles; -import java.util.List; -import java.util.Optional; -import java.util.concurrent.ExecutionException; - -/** - * A stand alone version of the Profiler that does not require a distributed - * execution environment like Apache Storm. - * - * <p>This class is used to create and manage profiles within the REPL environment. - */ -public class StandAloneProfiler { - - protected static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); - - /** - * The Stellar execution context. - */ - private Context context; - - /** - * The configuration for the Profiler. - */ - private ProfilerConfig config; - - /** - * The message router. - */ - private MessageRouter router; - - /** - * The message distributor. - */ - private MessageDistributor distributor; - - /** - * The factory that creates Clock objects. - */ - private ClockFactory clockFactory; - - /** - * Counts the number of messages that have been applied. - */ - private int messageCount; - - /** - * Counts the number of routes. - * - * If a message is not needed by any profiles, then there are 0 routes. - * If a message is needed by 1 profile then there is 1 route. - * If a message is needed by 2 profiles then there are 2 routes. - */ - private int routeCount; - - /** - * Create a new Profiler. - * - * @param config The Profiler configuration. - * @param periodDurationMillis The period duration in milliseconds. - * @param profileTimeToLiveMillis The time-to-live of a profile in milliseconds. - * @param maxNumberOfRoutes The max number of unique routes to maintain. After this is exceeded, lesser - * used routes will be evicted from the internal cache. - * @param context The Stellar execution context. - */ - public StandAloneProfiler(ProfilerConfig config, - long periodDurationMillis, - long profileTimeToLiveMillis, - long maxNumberOfRoutes, - Context context) { - this.context = context; - this.config = config; - this.router = new DefaultMessageRouter(context); - this.distributor = new DefaultMessageDistributor(periodDurationMillis, profileTimeToLiveMillis, maxNumberOfRoutes); - this.clockFactory = new DefaultClockFactory(); - this.messageCount = 0; - this.routeCount = 0; - } - - /** - * Apply a message to a set of profiles. - * @param message The message to apply. - */ - public void apply(JSONObject message) { - // route the message to the correct profile builders - List<MessageRoute> routes = router.route(message, config, context); - for (MessageRoute route : routes) { - distributor.distribute(route, context); - } - - routeCount += routes.size(); - messageCount += 1; - } - - /** - * Flush the set of profiles. - * @return A ProfileMeasurement for each (Profile, Entity) pair. - */ - public List<ProfileMeasurement> flush() { - return distributor.flush(); - } - - /** - * Returns the Profiler configuration. - * @return The Profiler configuration. - */ - public ProfilerConfig getConfig() { - return config; - } - - /** - * Returns the number of defined profiles. - * @return The number of defined profiles. - */ - public int getProfileCount() { - return (config == null) ? 0: config.getProfiles().size(); - } - - /** - * Returns the number of messages that have been applied. - * @return The number of messages that have been applied. - */ - public int getMessageCount() { - return messageCount; - } - - /** - * Returns the number of routes. - * @return The number of routes. - * @see MessageRoute - */ - public int getRouteCount() { - return routeCount; - } - - @Override - public String toString() { - return "Profiler{" + - getProfileCount() + " profile(s), " + - getMessageCount() + " messages(s), " + - getRouteCount() + " route(s)" + - '}'; - } -} http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-common/src/test/java/org/apache/metron/profiler/StandAloneProfilerTest.java ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-common/src/test/java/org/apache/metron/profiler/StandAloneProfilerTest.java b/metron-analytics/metron-profiler-common/src/test/java/org/apache/metron/profiler/StandAloneProfilerTest.java deleted file mode 100644 index 2269c86..0000000 --- a/metron-analytics/metron-profiler-common/src/test/java/org/apache/metron/profiler/StandAloneProfilerTest.java +++ /dev/null @@ -1,255 +0,0 @@ -/* - * - * 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; - -import org.adrianwalker.multilinestring.Multiline; -import org.apache.metron.common.configuration.profiler.ProfilerConfig; -import org.apache.metron.common.utils.JSONUtils; -import org.apache.metron.stellar.dsl.Context; -import org.json.simple.JSONObject; -import org.json.simple.parser.JSONParser; -import org.junit.Before; -import org.junit.Test; - -import java.io.ByteArrayInputStream; -import java.io.InputStream; -import java.util.Arrays; -import java.util.List; -import java.util.concurrent.TimeUnit; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - -/** - * Tests the StandAloneProfiler class. - */ -public class StandAloneProfilerTest { - - /** - * { - * "profiles": [ - * ] - * } - */ - @Multiline - private String noProfiles; - - /** - * { - * "profiles": [ - * { - * "profile": "profile1", - * "foreach": "'global'", - * "init": { "count": 0 }, - * "update": { "count": "count + 1" }, - * "result": "count" - * } - * ] - * } - */ - @Multiline - private String oneProfile; - - /** - * { - * "profiles": [ - * { - * "profile": "profile1", - * "foreach": "'global1'", - * "result": "'result'" - * }, - * { - * "profile": "profile2", - * "foreach": "'global2'", - * "result": "'result'" - * } - * ] - * } - */ - @Multiline - private String twoProfiles; - - /** - * { - * "ip_src_addr": "10.0.0.1", - * "ip_dst_addr": "10.0.0.20", - * "protocol": "HTTP", - * "timestamp": 2222222222222, - * } - */ - @Multiline - private String messageJson; - - private JSONObject message; - - private long periodDurationMillis = TimeUnit.MINUTES.toMillis(15); - - private Context context = Context.EMPTY_CONTEXT(); - - @Before - public void setup() throws Exception { - - // parse the input message - JSONParser parser = new JSONParser(); - message = (JSONObject) parser.parse(messageJson); - } - - @Test - public void testWithOneProfile() throws Exception { - - StandAloneProfiler profiler = createProfiler(oneProfile); - profiler.apply(message); - profiler.apply(message); - profiler.apply(message); - - List<ProfileMeasurement> measurements = profiler.flush(); - assertEquals(1, measurements.size()); - - // expect 1 measurement for the 1 profile that has been defined - ProfileMeasurement m = measurements.get(0); - assertEquals("profile1", m.getProfileName()); - assertEquals(3, m.getProfileValue()); - } - - - @Test - public void testWithTwoProfiles() throws Exception { - - StandAloneProfiler profiler = createProfiler(twoProfiles); - profiler.apply(message); - profiler.apply(message); - profiler.apply(message); - - List<ProfileMeasurement> measurements = profiler.flush(); - assertEquals(2, measurements.size()); - - // expect 2 measurements, 1 for each profile - List<String> expected = Arrays.asList(new String[] { "profile1", "profile2" }); - { - ProfileMeasurement m = measurements.get(0); - assertTrue(expected.contains(m.getProfileName())); - assertEquals("result", m.getProfileValue()); - } - { - ProfileMeasurement m = measurements.get(1); - assertTrue(expected.contains(m.getProfileName())); - assertEquals("result", m.getProfileValue()); - } - } - - /** - * The message count and route count will always be equal, if there is only one - * profile defined. The message count and route count can be different when there - * are multiple profiles defined that each use the same message. - */ - @Test - public void testRouteAndMessageCounters() throws Exception { - { - StandAloneProfiler profiler = createProfiler(noProfiles); - - profiler.apply(message); - assertEquals(1, profiler.getMessageCount()); - assertEquals(0, profiler.getRouteCount()); - - profiler.apply(message); - assertEquals(2, profiler.getMessageCount()); - assertEquals(0, profiler.getRouteCount()); - - profiler.apply(message); - assertEquals(3, profiler.getMessageCount()); - assertEquals(0, profiler.getRouteCount()); - } - { - StandAloneProfiler profiler = createProfiler(oneProfile); - - profiler.apply(message); - assertEquals(1, profiler.getMessageCount()); - assertEquals(1, profiler.getRouteCount()); - - profiler.apply(message); - assertEquals(2, profiler.getMessageCount()); - assertEquals(2, profiler.getRouteCount()); - - profiler.apply(message); - assertEquals(3, profiler.getMessageCount()); - assertEquals(3, profiler.getRouteCount()); - } - { - StandAloneProfiler profiler = createProfiler(twoProfiles); - - profiler.apply(message); - assertEquals(1, profiler.getMessageCount()); - assertEquals(2, profiler.getRouteCount()); - - profiler.apply(message); - assertEquals(2, profiler.getMessageCount()); - assertEquals(4, profiler.getRouteCount()); - - profiler.apply(message); - assertEquals(3, profiler.getMessageCount()); - assertEquals(6, profiler.getRouteCount()); - } - } - - @Test - public void testProfileCount() throws Exception { - { - StandAloneProfiler profiler = createProfiler(noProfiles); - assertEquals(0, profiler.getProfileCount()); - } - { - StandAloneProfiler profiler = createProfiler(oneProfile); - assertEquals(1, profiler.getProfileCount()); - } - { - StandAloneProfiler profiler = createProfiler(twoProfiles); - assertEquals(2, profiler.getProfileCount()); - } - } - - /** - * Creates a ProfilerConfig based on a string containing JSON. - * - * @param configAsJSON The config as JSON. - * @return The ProfilerConfig. - * @throws Exception - */ - private ProfilerConfig toProfilerConfig(String configAsJSON) throws Exception { - - InputStream in = new ByteArrayInputStream(configAsJSON.getBytes("UTF-8")); - return JSONUtils.INSTANCE.load(in, ProfilerConfig.class); - } - - /** - * Creates the StandAloneProfiler - * - * @param profileJson The Profiler configuration to use as a String containing JSON. - * @throws Exception - */ - private StandAloneProfiler createProfiler(String profileJson) throws Exception { - - // the TTL and max routes need not be bounded - long profileTimeToLiveMillis = Long.MAX_VALUE; - long maxNumberOfRoutes = Long.MAX_VALUE; - - ProfilerConfig config = toProfilerConfig(profileJson); - return new StandAloneProfiler(config, periodDurationMillis, profileTimeToLiveMillis, maxNumberOfRoutes, context); - } -} http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-repl/README.md ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-repl/README.md b/metron-analytics/metron-profiler-repl/README.md new file mode 100644 index 0000000..e0f61d4 --- /dev/null +++ b/metron-analytics/metron-profiler-repl/README.md @@ -0,0 +1,178 @@ +<!-- +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. +--> +# Metron Profiler for the Stellar REPL + +This project allows profiles to be executed within the Stellar REPL. This is a port of the Profiler to run in the Stellar REPL. + +* [Introduction](#introduction) +* [Getting Started](#getting-started) +* [Installation](#installation) + +## Introduction + +Creating and refining profiles is an iterative process. Iterating against a live stream of data is slow, difficult and error prone. Running the Profiler in the Stellar REPL provides a controlled and isolated execution environment to create, refine and troubleshoot profiles. + +For an introduction to the Profiler, see the [Profiler README](../metron-profiler/README.md). + +## Getting Started + +This section describes how to get started using the Profiler in the Stellar REPL. This outlines a useful workflow for creating, testing, and deploying profiles. + +1. Launch the Stellar REPL. + ``` + [root@node1 ~]# $METRON_HOME/bin/stellar + Stellar, Go! + [Stellar]>>> + ``` + +1. The following functions should be accessible. Documentation is also provided for each function; for example by executing`?PROFILER_FLUSH`. + ``` + [Stellar]>>> %functions PROFILER + PROFILER_APPLY, PROFILER_FLUSH, PROFILER_INIT + ``` + +1. Create a simple `hello-world` profile that will count the number of messages for each `ip_src_addr`. The `SHELL_EDIT` function will open an editor in which you can copy/paste the following Profiler configuration. + ``` + [Stellar]>>> conf := SHELL_EDIT() + [Stellar]>>> conf + { + "profiles": [ + { + "profile": "hello-world", + "foreach": "ip_src_addr", + "init": { "count": "0" }, + "update": { "count": "count + 1" }, + "result": "count" + } + ] + } + ``` + +1. Create the profile execution environment. + + The Profiler will output the number of profiles that have been defined, the number of messages that have been applied and the number of routes that have been followed. + + While the idea of a profile and message may be well understood, a route may need further explanation. A route is defined when a message is applied to a specific profile. + * If a message is not needed by any profile, then there are no routes. + * If a message is needed by one profile, then one route has been followed. + * If a message is needed by two profiles, then two routes have been followed. + + ``` + [Stellar]>>> profiler := PROFILER_INIT(conf) + [Stellar]>>> profiler + Profiler{1 profile(s), 0 messages(s), 0 route(s)} + ``` + +1. Create a message that mimics the telemetry that your profile will consume. + + This message can be as simple or complex as you like. For the `hello-world` profile, all you need is a message containing an `ip_src_addr` field. + + ``` + [Stellar]>>> msg := SHELL_EDIT() + [Stellar]>>> msg + { + "ip_src_addr": "10.0.0.1" + } + ``` + +1. Apply the message to your Profiler, as many times as you like. + + ``` + [Stellar]>>> PROFILER_APPLY(msg, profiler) + Profiler{1 profile(s), 1 messages(s), 1 route(s)} + ``` + ``` + [Stellar]>>> PROFILER_APPLY(msg, profiler) + Profiler{1 profile(s), 2 messages(s), 2 route(s)} + ``` + ``` + [Stellar]>>> PROFILER_APPLY(msg, profiler) + Profiler{1 profile(s), 3 messages(s), 3 route(s)} + ``` + +1. Flush the Profiler. + + A flush is what occurs at the end of each 15 minute period in the Profiler. The result is a list of Profile Measurements. Each measurement is a map containing detailed information about the profile data that has been generated. The `value` field is what is written to HBase when running the Profiler in either Storm or Spark. + + There will always be one measurement for each [profile, entity] pair. This profile simply counts the number of messages by IP source address. Notice that the value is '3' for the entity '10.0.0.1' as we applied 3 messages with an 'ip_src_addr' of â10.0.0.1'. + + ``` + [Stellar]>>> values := PROFILER_FLUSH(profiler) + [Stellar]>>> values + [{period={duration=900000, period=1669628, start=1502665200000, end=1502666100000}, + profile=hello-world, groups=[], value=3, entity=10.0.0.1}] + ``` + +1. In addition to testing with mock data, you can also apply real, live telemetry to your profile. This can be useful to test your profile against the complexities that exist in real data. + + This example extracts 10 messages of live, enriched telemetry to test your profile(s). + ``` + [Stellar]>>> msgs := KAFKA_GET("indexing", 10) + [Stellar]>>> LENGTH(msgs) + 10 + ``` + Now apply those 10 messages to your profile. + ``` + [Stellar]>>> PROFILER_APPLY(msgs, profiler) + Profiler{1 profile(s), 10 messages(s), 10 route(s)} + ``` + +1. After you are satisfied with your profile, the next step is to deploy the profile against the live stream of telemetry being capture by Metron. This involves deploying the profile to either the [Storm Profiler](../metron-profiler/README.md) or the [Spark Profiler](../metron-profiler-spark/README.md). + + +## Installation + +This package is installed automatically when installing Metron using the Ambari MPack. + +This package can also be installed manually by using either the RPM or DEB package. + +### Build the RPM + +1. Build Metron. + ``` + mvn clean package -DskipTests -T2C + ``` + +1. Build the RPMs. + ``` + cd metron-deployment/ + mvn clean package -Pbuild-rpms + ``` + +1. Retrieve the package. + ``` + find ./ -name "metron-profiler-repl*.rpm" + ``` + +### Build the DEB + +1. Build Metron. + ``` + mvn clean package -DskipTests -T2C + ``` + +1. Build the DEBs. + ``` + cd metron-deployment/ + mvn clean package -Pbuild-debs + ``` + +1. Retrieve the package. + ``` + find ./ -name "metron-profiler-repl*.deb" + ``` http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-repl/pom.xml ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-repl/pom.xml b/metron-analytics/metron-profiler-repl/pom.xml new file mode 100644 index 0000000..2447b40 --- /dev/null +++ b/metron-analytics/metron-profiler-repl/pom.xml @@ -0,0 +1,150 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ 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. + ~ + --> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <artifactId>metron-analytics</artifactId> + <groupId>org.apache.metron</groupId> + <version>0.5.1</version> + </parent> + <artifactId>metron-profiler-repl</artifactId> + <url>https://metron.apache.org/</url> + <properties> + <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> + <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> + </properties> + <dependencies> + <dependency> + <groupId>org.apache.metron</groupId> + <artifactId>metron-profiler-common</artifactId> + <version>${project.parent.version}</version> + </dependency> + <dependency> + <groupId>org.apache.metron</groupId> + <artifactId>metron-profiler-client</artifactId> + <version>${project.parent.version}</version> + </dependency> + <dependency> + <groupId>org.apache.metron</groupId> + <artifactId>metron-common</artifactId> + <version>${project.parent.version}</version> + <exclusions> + <exclusion> + <groupId>com.google.guava</groupId> + <artifactId>guava</artifactId> + </exclusion> + <exclusion> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <!-- allows profiles to use the Stellar stats functions --> + <groupId>org.apache.metron</groupId> + <artifactId>metron-statistics</artifactId> + <version>${project.parent.version}</version> + <exclusions> + <exclusion> + <artifactId>kryo</artifactId> + <groupId>com.esotericsoftware</groupId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.logging.log4j</groupId> + <artifactId>log4j-api</artifactId> + <version>${global_log4j_core_version}</version> + <scope>test</scope> + </dependency> + <dependency> + <groupId>org.apache.logging.log4j</groupId> + <artifactId>log4j-core</artifactId> + <version>${global_log4j_core_version}</version> + <scope>test</scope> + </dependency> + </dependencies> + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-shade-plugin</artifactId> + <version>${global_shade_version}</version> + <configuration> + <createDependencyReducedPom>true</createDependencyReducedPom> + </configuration> + <executions> + <execution> + <phase>package</phase> + <goals> + <goal>shade</goal> + </goals> + <configuration> + <filters> + <filter> + <artifact>*:*</artifact> + <excludes> + <exclude>META-INF/*.SF</exclude> + <exclude>META-INF/*.DSA</exclude> + <exclude>META-INF/*.RSA</exclude> + </excludes> + </filter> + </filters> + <transformers> + <transformer + implementation="org.apache.maven.plugins.shade.resource.DontIncludeResourceTransformer"> + <resources> + <resource>.yaml</resource> + <resource>LICENSE.txt</resource> + <resource>ASL2.0</resource> + <resource>NOTICE.txt</resource> + </resources> + </transformer> + <transformer + implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> + <transformer + implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> + <mainClass></mainClass> + </transformer> + </transformers> + </configuration> + </execution> + </executions> + </plugin> + <plugin> + <artifactId>maven-assembly-plugin</artifactId> + <configuration> + <descriptor>src/main/assembly/assembly.xml</descriptor> + </configuration> + <executions> + <execution> + <id>make-assembly</id> <!-- this is used for inheritance merges --> + <phase>package</phase> <!-- bind to the packaging phase --> + <goals> + <goal>single</goal> + </goals> + </execution> + </executions> + </plugin> + </plugins> + </build> +</project> http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-repl/src/main/assembly/assembly.xml ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-repl/src/main/assembly/assembly.xml b/metron-analytics/metron-profiler-repl/src/main/assembly/assembly.xml new file mode 100644 index 0000000..fa6f0c1 --- /dev/null +++ b/metron-analytics/metron-profiler-repl/src/main/assembly/assembly.xml @@ -0,0 +1,36 @@ +<!-- + ~ 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. + ~ + --> + +<assembly> + <id>archive</id> + <formats> + <format>tar.gz</format> + </formats> + <includeBaseDirectory>false</includeBaseDirectory> + <fileSets> + <fileSet> + <directory>${project.basedir}/target</directory> + <includes> + <include>${project.artifactId}-${project.version}.jar</include> + </includes> + <outputDirectory>lib</outputDirectory> + <useDefaultExcludes>true</useDefaultExcludes> + </fileSet> + </fileSets> +</assembly> http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-repl/src/main/java/org/apache/metron/profiler/repl/ProfilerFunctions.java ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-repl/src/main/java/org/apache/metron/profiler/repl/ProfilerFunctions.java b/metron-analytics/metron-profiler-repl/src/main/java/org/apache/metron/profiler/repl/ProfilerFunctions.java new file mode 100644 index 0000000..506c47e --- /dev/null +++ b/metron-analytics/metron-profiler-repl/src/main/java/org/apache/metron/profiler/repl/ProfilerFunctions.java @@ -0,0 +1,278 @@ +/* + * 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.repl; + +import org.apache.commons.lang3.ClassUtils; +import org.apache.metron.common.configuration.profiler.ProfilerConfig; +import org.apache.metron.common.utils.JSONUtils; +import org.apache.metron.profiler.ProfileMeasurement; +import org.apache.metron.profiler.client.stellar.Util; +import org.apache.metron.stellar.dsl.Context; +import org.apache.metron.stellar.dsl.ParseException; +import org.apache.metron.stellar.dsl.Stellar; +import org.apache.metron.stellar.dsl.StellarFunction; +import org.json.simple.JSONArray; +import org.json.simple.JSONObject; +import org.json.simple.parser.JSONParser; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.lang.invoke.MethodHandles; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import static java.lang.String.format; +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.stellar.dsl.Context.Capabilities.GLOBAL_CONFIG; + +/** + * Stellar functions that allow interaction with the core Profiler components + * through the Stellar REPL. + */ +public class ProfilerFunctions { + + private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + @Stellar( + namespace="PROFILER", + name="INIT", + description="Creates a local profile runner that can execute profiles.", + params={ + "config", "The profiler configuration as a string." + }, + returns="A local profile runner." + ) + public static class ProfilerInit implements StellarFunction { + + @Override + public void initialize(Context context) { + } + + @Override + public boolean isInitialized() { + return true; + } + + @Override + public Object apply(List<Object> args, Context context) { + @SuppressWarnings("unchecked") + Map<String, Object> global = (Map<String, Object>) context.getCapability(GLOBAL_CONFIG, false) + .orElse(Collections.emptyMap()); + + // how long is the profile period? + long duration = PROFILER_PERIOD.getOrDefault(global, PROFILER_PERIOD.getDefault(), Long.class); + String configuredUnits = PROFILER_PERIOD_UNITS.getOrDefault(global, PROFILER_PERIOD_UNITS.getDefault(), String.class); + long periodDurationMillis = TimeUnit.valueOf(configuredUnits).toMillis(duration); + + // user must provide the configuration for the profiler + String arg0 = Util.getArg(0, String.class, args); + ProfilerConfig profilerConfig; + try { + profilerConfig = JSONUtils.INSTANCE.load(arg0, ProfilerConfig.class); + + } catch(IOException e) { + throw new IllegalArgumentException("Invalid profiler configuration", e); + } + + // the TTL and max routes do not matter here + long profileTimeToLiveMillis = Long.MAX_VALUE; + long maxNumberOfRoutes = Long.MAX_VALUE; + return new StandAloneProfiler(profilerConfig, periodDurationMillis, profileTimeToLiveMillis, maxNumberOfRoutes, context); + } + } + + @Stellar( + namespace="PROFILER", + name="APPLY", + description="Apply a message to a local profile runner.", + params={ + "message(s)", "The message to apply; a JSON string or list of JSON strings.", + "profiler", "A local profile runner returned by PROFILER_INIT." + }, + returns="The local profile runner." + ) + public static class ProfilerApply implements StellarFunction { + + private JSONParser parser; + + @Override + public void initialize(Context context) { + parser = new JSONParser(); + } + + @Override + public boolean isInitialized() { + return parser != null; + } + + @Override + public Object apply(List<Object> args, Context context) throws ParseException { + + // the use can pass in one or more messages in a few different forms + Object arg0 = Util.getArg(0, Object.class, args); + List<JSONObject> messages = getMessages(arg0); + + // user must provide the stand alone profiler + StandAloneProfiler profiler = Util.getArg(1, StandAloneProfiler.class, args); + for (JSONObject message : messages) { + profiler.apply(message); + } + + return profiler; + } + + /** + * Gets a message or messages from the function arguments. + * + * @param arg The function argument containing the message(s). + * @return A list of messages + */ + private List<JSONObject> getMessages(Object arg) { + List<JSONObject> messages; + + if (arg instanceof String) { + messages = getMessagesFromString((String) arg); + + } else if (arg instanceof Iterable) { + messages = getMessagesFromIterable((Iterable<String>) arg); + + } else if (arg instanceof JSONObject) { + messages = Collections.singletonList((JSONObject) arg); + + } else { + throw new IllegalArgumentException(format("invalid message: found '%s', expected String, List, or JSONObject", + ClassUtils.getShortClassName(arg, "null"))); + } + + return messages; + } + + /** + * Gets a message or messages from a List + * + * @param strings The function argument that is a bunch of strings. + * @return A list of messages. + */ + private List<JSONObject> getMessagesFromIterable(Iterable<String> strings) { + List<JSONObject> messages = new ArrayList<>(); + + // the user pass in a list of strings + for (String str : strings) { + messages.addAll(getMessagesFromString(str)); + } + + return messages; + } + + /** + * Gets a message or messages from a String argument. + * + * @param arg0 The function argument is just a List. + * @return A list of messages. + */ + private List<JSONObject> getMessagesFromString(String arg0) { + List<JSONObject> messages = new ArrayList<>(); + + try { + Object parsedArg0 = parser.parse(arg0); + if (parsedArg0 instanceof JSONObject) { + // if the string only contains one message + messages.add((JSONObject) parsedArg0); + + } else if (parsedArg0 instanceof JSONArray) { + // if the string contains multiple messages + JSONArray jsonArray = (JSONArray) parsedArg0; + for (Object item : jsonArray) { + messages.addAll(getMessages(item)); + } + + } else { + throw new IllegalArgumentException(format("invalid message: found '%s', expected JSONObject or JSONArray", + ClassUtils.getShortClassName(parsedArg0, "null"))); + } + + } catch (org.json.simple.parser.ParseException e) { + throw new IllegalArgumentException(format("invalid message: '%s'", e.getMessage()), e); + } + + return messages; + } + } + + @Stellar( + namespace="PROFILER", + name="FLUSH", + description="Flush a local profile runner.", + params={ + "profiler", "A local profile runner returned by PROFILER_INIT." + }, + returns="A list of the profile values." + ) + public static class ProfilerFlush implements StellarFunction { + + @Override + public void initialize(Context context) { + } + + @Override + public boolean isInitialized() { + return true; + } + + @Override + public Object apply(List<Object> args, Context context) throws ParseException { + + // user must provide the stand-alone profiler + StandAloneProfiler profiler = Util.getArg(0, StandAloneProfiler.class, args); + if(profiler == null) { + throw new IllegalArgumentException(format("expected the profiler returned by PROFILER_INIT, found null")); + } + + // transform the profile measurements into maps to simplify manipulation in stellar + List<Map<String, Object>> measurements = new ArrayList<>(); + for(ProfileMeasurement m : profiler.flush()) { + + // create a map for the profile period + Map<String, Object> period = new HashMap<>(); + period.put("period", m.getPeriod().getPeriod()); + period.put("start", m.getPeriod().getStartTimeMillis()); + period.put("duration", m.getPeriod().getDurationMillis()); + period.put("end", m.getPeriod().getEndTimeMillis()); + + // create a map for the measurement + Map<String, Object> measurement = new HashMap<>(); + measurement.put("profile", m.getProfileName()); + measurement.put("entity", m.getEntity()); + measurement.put("value", m.getProfileValue()); + measurement.put("groups", m.getGroups()); + measurement.put("period", period); + + measurements.add(measurement); + } + + return measurements; + } + } +} http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-repl/src/main/java/org/apache/metron/profiler/repl/StandAloneProfiler.java ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-repl/src/main/java/org/apache/metron/profiler/repl/StandAloneProfiler.java b/metron-analytics/metron-profiler-repl/src/main/java/org/apache/metron/profiler/repl/StandAloneProfiler.java new file mode 100644 index 0000000..1206a99 --- /dev/null +++ b/metron-analytics/metron-profiler-repl/src/main/java/org/apache/metron/profiler/repl/StandAloneProfiler.java @@ -0,0 +1,176 @@ +/* + * 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.repl; + +import org.apache.metron.common.configuration.profiler.ProfilerConfig; +import org.apache.metron.profiler.DefaultMessageDistributor; +import org.apache.metron.profiler.DefaultMessageRouter; +import org.apache.metron.profiler.MessageDistributor; +import org.apache.metron.profiler.MessageRoute; +import org.apache.metron.profiler.MessageRouter; +import org.apache.metron.profiler.ProfileMeasurement; +import org.apache.metron.profiler.clock.ClockFactory; +import org.apache.metron.profiler.clock.DefaultClockFactory; +import org.apache.metron.stellar.dsl.Context; +import org.json.simple.JSONObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.lang.invoke.MethodHandles; +import java.util.List; + +/** + * A stand alone version of the Profiler that does not require a distributed + * execution environment like Apache Storm. + * + * <p>This class is used to create and manage profiles within the REPL environment. + */ +public class StandAloneProfiler { + + protected static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + /** + * The Stellar execution context. + */ + private Context context; + + /** + * The configuration for the Profiler. + */ + private ProfilerConfig config; + + /** + * The message router. + */ + private MessageRouter router; + + /** + * The message distributor. + */ + private MessageDistributor distributor; + + /** + * The factory that creates Clock objects. + */ + private ClockFactory clockFactory; + + /** + * Counts the number of messages that have been applied. + */ + private int messageCount; + + /** + * Counts the number of routes. + * + * If a message is not needed by any profiles, then there are 0 routes. + * If a message is needed by 1 profile then there is 1 route. + * If a message is needed by 2 profiles then there are 2 routes. + */ + private int routeCount; + + /** + * Create a new Profiler. + * + * @param config The Profiler configuration. + * @param periodDurationMillis The period duration in milliseconds. + * @param profileTimeToLiveMillis The time-to-live of a profile in milliseconds. + * @param maxNumberOfRoutes The max number of unique routes to maintain. After this is exceeded, lesser + * used routes will be evicted from the internal cache. + * @param context The Stellar execution context. + */ + public StandAloneProfiler(ProfilerConfig config, + long periodDurationMillis, + long profileTimeToLiveMillis, + long maxNumberOfRoutes, + Context context) { + this.context = context; + this.config = config; + this.router = new DefaultMessageRouter(context); + this.distributor = new DefaultMessageDistributor(periodDurationMillis, profileTimeToLiveMillis, maxNumberOfRoutes); + this.clockFactory = new DefaultClockFactory(); + this.messageCount = 0; + this.routeCount = 0; + } + + /** + * Apply a message to a set of profiles. + * @param message The message to apply. + */ + public void apply(JSONObject message) { + // route the message to the correct profile builders + List<MessageRoute> routes = router.route(message, config, context); + for (MessageRoute route : routes) { + distributor.distribute(route, context); + } + + routeCount += routes.size(); + messageCount += 1; + } + + /** + * Flush the set of profiles. + * @return A ProfileMeasurement for each (Profile, Entity) pair. + */ + public List<ProfileMeasurement> flush() { + return distributor.flush(); + } + + /** + * Returns the Profiler configuration. + * @return The Profiler configuration. + */ + public ProfilerConfig getConfig() { + return config; + } + + /** + * Returns the number of defined profiles. + * @return The number of defined profiles. + */ + public int getProfileCount() { + return (config == null) ? 0: config.getProfiles().size(); + } + + /** + * Returns the number of messages that have been applied. + * @return The number of messages that have been applied. + */ + public int getMessageCount() { + return messageCount; + } + + /** + * Returns the number of routes. + * @return The number of routes. + * @see MessageRoute + */ + public int getRouteCount() { + return routeCount; + } + + @Override + public String toString() { + return "Profiler{" + + getProfileCount() + " profile(s), " + + getMessageCount() + " messages(s), " + + getRouteCount() + " route(s)" + + '}'; + } +} http://git-wip-us.apache.org/repos/asf/metron/blob/9455c4ee/metron-analytics/metron-profiler-repl/src/test/java/org/apache/metron/profiler/repl/ProfilerFunctionsTest.java ---------------------------------------------------------------------- diff --git a/metron-analytics/metron-profiler-repl/src/test/java/org/apache/metron/profiler/repl/ProfilerFunctionsTest.java b/metron-analytics/metron-profiler-repl/src/test/java/org/apache/metron/profiler/repl/ProfilerFunctionsTest.java new file mode 100644 index 0000000..6cf34b0 --- /dev/null +++ b/metron-analytics/metron-profiler-repl/src/test/java/org/apache/metron/profiler/repl/ProfilerFunctionsTest.java @@ -0,0 +1,332 @@ +/* + * 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.repl; + +import org.adrianwalker.multilinestring.Multiline; +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.ParseException; +import org.apache.metron.stellar.dsl.functions.resolver.SimpleFunctionResolver; +import org.json.simple.JSONObject; +import org.json.simple.parser.JSONParser; +import org.junit.Before; +import org.junit.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +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.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; + +/** + * Tests the ProfilerFunctions class. + */ +public class ProfilerFunctionsTest { + + /** + * { + * "ip_src_addr": "10.0.0.1", + * "ip_dst_addr": "10.0.0.2", + * "source.type": "test", + * } + */ + @Multiline + private String message; + + /** + * [ + * { + * "ip_src_addr": "10.0.0.1", + * "ip_dst_addr": "10.0.0.2", + * "source.type": "test", + * }, + * { + * "ip_src_addr": "10.0.0.1", + * "ip_dst_addr": "10.0.0.2", + * "source.type": "test", + * }, + * { + * "ip_src_addr": "10.0.0.1", + * "ip_dst_addr": "10.0.0.2", + * "source.type": "test", + * } + * ] + */ + @Multiline + private String messages; + + /** + * { + * "profiles": [ + * { + * "profile": "hello-world", + * "foreach": "ip_src_addr", + * "init": { "count": 0 }, + * "update": { "count": "count + 1" }, + * "result": "count" + * } + * ] + * } + */ + @Multiline + private String helloWorldProfilerDef; + + private static final long periodDuration = 15; + private static final String periodUnits = "MINUTES"; + private StellarStatefulExecutor executor; + private Map<String, Object> state; + + private <T> T run(String expression, Class<T> clazz) { + return executor.execute(expression, state, clazz); + } + + @Before + public void setup() { + state = new HashMap<>(); + + // global properties + Map<String, Object> global = new HashMap<String, Object>() {{ + put(PROFILER_PERIOD.getKey(), Long.toString(periodDuration)); + put(PROFILER_PERIOD_UNITS.getKey(), periodUnits.toString()); + }}; + + // create the stellar execution environment + executor = new DefaultStellarStatefulExecutor( + new SimpleFunctionResolver() + .withClass(ProfilerFunctions.ProfilerInit.class) + .withClass(ProfilerFunctions.ProfilerApply.class) + .withClass(ProfilerFunctions.ProfilerFlush.class), + new Context.Builder() + .with(Context.Capabilities.GLOBAL_CONFIG, () -> global) + .build()); + } + + @Test + public void testProfilerInitNoProfiles() { + state.put("config", "{ \"profiles\" : [] }"); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + assertNotNull(profiler); + assertEquals(0, profiler.getProfileCount()); + assertEquals(0, profiler.getMessageCount()); + assertEquals(0, profiler.getRouteCount()); + } + + @Test + public void testProfilerInitWithProfiles() { + state.put("config", helloWorldProfilerDef); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + assertNotNull(profiler); + assertEquals(1, profiler.getProfileCount()); + assertEquals(0, profiler.getMessageCount()); + assertEquals(0, profiler.getRouteCount()); + } + + @Test(expected = ParseException.class) + public void testProfilerInitNoArgs() { + run("PROFILER_INIT()", StandAloneProfiler.class); + } + + @Test(expected = ParseException.class) + public void testProfilerInitInvalidArg() { + run("PROFILER_INIT({ \"invalid\": 2 })", StandAloneProfiler.class); + } + + @Test + public void testProfilerInitWithNoGlobalConfig() { + state.put("config", helloWorldProfilerDef); + String expression = "PROFILER_INIT(config)"; + + // use an executor with no GLOBAL_CONFIG defined in the context + StellarStatefulExecutor executor = new DefaultStellarStatefulExecutor( + new SimpleFunctionResolver() + .withClass(ProfilerFunctions.ProfilerInit.class) + .withClass(ProfilerFunctions.ProfilerApply.class) + .withClass(ProfilerFunctions.ProfilerFlush.class), + Context.EMPTY_CONTEXT()); + StandAloneProfiler profiler = executor.execute(expression, state, StandAloneProfiler.class); + + assertNotNull(profiler); + assertEquals(1, profiler.getProfileCount()); + assertEquals(0, profiler.getMessageCount()); + assertEquals(0, profiler.getRouteCount()); + } + + @Test + public void testProfilerApplyWithString() { + + // initialize the profiler + state.put("config", helloWorldProfilerDef); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + state.put("profiler", profiler); + + // apply a message to the profiler + state.put("message", message); + StandAloneProfiler result = run("PROFILER_APPLY(message, profiler)", StandAloneProfiler.class); + + // validate + assertSame(profiler, result); + assertEquals(1, profiler.getProfileCount()); + assertEquals(1, profiler.getMessageCount()); + assertEquals(1, profiler.getRouteCount()); + } + + @Test + public void testProfilerApplyWithJSONObject() throws Exception { + + // initialize the profiler + state.put("config", helloWorldProfilerDef); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + state.put("profiler", profiler); + + // apply a message to the profiler + JSONParser parser = new JSONParser(); + JSONObject jsonObject = (JSONObject) parser.parse(message); + state.put("jsonObj", jsonObject); + StandAloneProfiler result = run("PROFILER_APPLY(jsonObj, profiler)", StandAloneProfiler.class); + + // validate + assertSame(profiler, result); + assertEquals(1, profiler.getProfileCount()); + assertEquals(1, profiler.getMessageCount()); + assertEquals(1, profiler.getRouteCount()); + } + + @Test + public void testProfilerApplyWithMultipleMessagesInJSONString() { + + // initialize the profiler + state.put("config", helloWorldProfilerDef); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + state.put("profiler", profiler); + + // apply a message to the profiler + state.put("messages", messages); + StandAloneProfiler result = run("PROFILER_APPLY(messages, profiler)", StandAloneProfiler.class); + + // validate + assertSame(profiler, result); + assertEquals(1, profiler.getProfileCount()); + assertEquals(3, profiler.getMessageCount()); + assertEquals(3, profiler.getRouteCount()); + } + + @Test + public void testProfilerApplyWithListOfMessages() { + + // initialize the profiler + state.put("config", helloWorldProfilerDef); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + state.put("profiler", profiler); + + // apply a message to the profiler + state.put("msg", message); + StandAloneProfiler result = run("PROFILER_APPLY([msg, msg, msg], profiler)", StandAloneProfiler.class); + + // validate + assertSame(profiler, result); + assertEquals(1, profiler.getProfileCount()); + assertEquals(3, profiler.getMessageCount()); + assertEquals(3, profiler.getRouteCount()); + } + + + @Test + public void testProfilerApplyWithEmptyList() { + + // initialize the profiler + state.put("config", helloWorldProfilerDef); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + state.put("profiler", profiler); + + // apply a message to the profiler + state.put("messages", "[ ]"); + StandAloneProfiler result = run("PROFILER_APPLY(messages, profiler)", StandAloneProfiler.class); + + // validate + assertSame(profiler, result); + assertEquals(1, profiler.getProfileCount()); + assertEquals(0, profiler.getMessageCount()); + assertEquals(0, profiler.getRouteCount()); + } + + @Test(expected = ParseException.class) + public void testProfilerApplyWithNoArgs() { + run("PROFILER_APPLY()", StandAloneProfiler.class); + } + + @Test(expected = ParseException.class) + public void testProfilerApplyWithInvalidArg() { + run("PROFILER_APPLY(undefined)", StandAloneProfiler.class); + } + + @Test(expected = ParseException.class) + public void testProfilerApplyWithNullMessage() { + + // initialize the profiler + state.put("config", helloWorldProfilerDef); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + state.put("profiler", profiler); + + // there is no 'messages' variable - should throw exception + run("PROFILER_APPLY(messages, profiler)", StandAloneProfiler.class); + } + + @Test + public void testProfilerFlush() { + + // initialize the profiler + state.put("config", helloWorldProfilerDef); + StandAloneProfiler profiler = run("PROFILER_INIT(config)", StandAloneProfiler.class); + state.put("profiler", profiler); + + // apply a message to the profiler + state.put("message", message); + run("PROFILER_APPLY(message, profiler)", StandAloneProfiler.class); + + // flush the profiles + List<Map<String, Object>> measurements = run("PROFILER_FLUSH(profiler)", List.class); + + // validate + assertNotNull(measurements); + assertEquals(1, measurements.size()); + + Map<String, Object> measurement = measurements.get(0); + assertEquals("hello-world", measurement.get("profile")); + assertEquals("10.0.0.1", measurement.get("entity")); + assertEquals(1, measurement.get("value")); + assertEquals(Collections.emptyList(), measurement.get("groups")); + } + + @Test(expected = ParseException.class) + public void testProfilerFlushNoArgs() { + run("PROFILER_FLUSH()", StandAloneProfiler.class); + } + + @Test(expected = ParseException.class) + public void testProfilerFlushInvalidArg() { + run("PROFILER_FLUSH(undefined)", StandAloneProfiler.class); + } +}
