| /** |
| * 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.storm.daemon; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.TreeMap; |
| import org.apache.storm.Config; |
| import org.apache.storm.Constants; |
| import org.apache.storm.Thrift; |
| import org.apache.storm.generated.Bolt; |
| import org.apache.storm.generated.ComponentCommon; |
| import org.apache.storm.generated.GlobalStreamId; |
| import org.apache.storm.generated.Grouping; |
| import org.apache.storm.generated.InvalidTopologyException; |
| import org.apache.storm.generated.NodeInfo; |
| import org.apache.storm.generated.SpoutSpec; |
| import org.apache.storm.generated.StateSpoutSpec; |
| import org.apache.storm.generated.StormTopology; |
| import org.apache.storm.generated.StreamInfo; |
| import org.apache.storm.metric.EventLoggerBolt; |
| import org.apache.storm.metric.MetricsConsumerBolt; |
| import org.apache.storm.metric.SystemBolt; |
| import org.apache.storm.metric.filter.FilterByMetricName; |
| import org.apache.storm.metric.util.DataPointExpander; |
| import org.apache.storm.security.auth.IAuthorizer; |
| import org.apache.storm.shade.org.apache.commons.lang.StringUtils; |
| import org.apache.storm.shade.org.json.simple.JSONValue; |
| import org.apache.storm.task.IBolt; |
| import org.apache.storm.task.WorkerTopologyContext; |
| import org.apache.storm.tuple.Fields; |
| import org.apache.storm.utils.ConfigUtils; |
| import org.apache.storm.utils.ObjectReader; |
| import org.apache.storm.utils.Utils; |
| import org.apache.storm.utils.WrappedInvalidTopologyException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| public class StormCommon { |
| public static final String SYSTEM_STREAM_ID = "__system"; |
| public static final String EVENTLOGGER_COMPONENT_ID = "__eventlogger"; |
| public static final String EVENTLOGGER_STREAM_ID = "__eventlog"; |
| public static final String TOPOLOGY_METRICS_CONSUMER_CLASS = "class"; |
| public static final String TOPOLOGY_METRICS_CONSUMER_ARGUMENT = "argument"; |
| public static final String TOPOLOGY_METRICS_CONSUMER_MAX_RETAIN_METRIC_TUPLES = "max.retain.metric.tuples"; |
| public static final String TOPOLOGY_METRICS_CONSUMER_PARALLELISM_HINT = "parallelism.hint"; |
| public static final String TOPOLOGY_METRICS_CONSUMER_WHITELIST = "whitelist"; |
| public static final String TOPOLOGY_METRICS_CONSUMER_BLACKLIST = "blacklist"; |
| public static final String TOPOLOGY_METRICS_CONSUMER_EXPAND_MAP_TYPE = "expandMapType"; |
| public static final String TOPOLOGY_METRICS_CONSUMER_METRIC_NAME_SEPARATOR = "metricNameSeparator"; |
| public static final String TOPOLOGY_EVENT_LOGGER_CLASS = "class"; |
| public static final String TOPOLOGY_EVENT_LOGGER_ARGUMENTS = "arguments"; |
| private static final Logger LOG = LoggerFactory.getLogger(StormCommon.class); |
| // A singleton instance allows us to mock delegated static methods in our |
| // tests by subclassing. |
| private static StormCommon _instance = new StormCommon(); |
| |
| /** |
| * Provide an instance of this class for delegates to use. To mock out delegated methods, provide an instance of a subclass that |
| * overrides the implementation of the delegated method. |
| * |
| * @param common a StormCommon instance |
| * @return the previously set instance |
| */ |
| public static StormCommon setInstance(StormCommon common) { |
| StormCommon oldInstance = _instance; |
| _instance = common; |
| return oldInstance; |
| } |
| |
| public static void validateDistributedMode(Map<String, Object> conf) { |
| if (ConfigUtils.isLocalMode(conf)) { |
| throw new IllegalArgumentException("Cannot start server in local mode!"); |
| } |
| } |
| |
| private static Set<String> validateIds(Map<String, ? extends Object> componentMap) throws InvalidTopologyException { |
| Set<String> keys = componentMap.keySet(); |
| for (String id : keys) { |
| if (Utils.isSystemId(id)) { |
| throw new WrappedInvalidTopologyException(id + " is not a valid component id."); |
| } |
| } |
| for (Object componentObj : componentMap.values()) { |
| ComponentCommon common = getComponentCommon(componentObj); |
| Set<String> streamIds = common.get_streams().keySet(); |
| for (String id : streamIds) { |
| if (Utils.isSystemId(id)) { |
| throw new WrappedInvalidTopologyException(id + " is not a valid stream id."); |
| } |
| } |
| } |
| return keys; |
| } |
| |
| private static void validateIds(StormTopology topology) throws InvalidTopologyException { |
| List<String> componentIds = new ArrayList<>(); |
| componentIds.addAll(validateIds(topology.get_bolts())); |
| componentIds.addAll(validateIds(topology.get_spouts())); |
| componentIds.addAll(validateIds(topology.get_state_spouts())); |
| |
| List<String> offending = Utils.getRepeat(componentIds); |
| if (!offending.isEmpty()) { |
| throw new WrappedInvalidTopologyException("Duplicate component ids: " + offending); |
| } |
| } |
| |
| private static boolean isEmptyInputs(ComponentCommon common) { |
| if (common.get_inputs() == null) { |
| return true; |
| } else { |
| return common.get_inputs().isEmpty(); |
| } |
| } |
| |
| public static Map<String, Object> allComponents(StormTopology topology) { |
| Map<String, Object> components = new HashMap<>(topology.get_bolts()); |
| components.putAll(topology.get_spouts()); |
| components.putAll(topology.get_state_spouts()); |
| return components; |
| } |
| |
| @SuppressWarnings("unchecked") |
| public static Map<String, Object> componentConf(Object component) { |
| try { |
| Map<String, Object> conf = new HashMap<>(); |
| ComponentCommon common = getComponentCommon(component); |
| String jconf = common.get_json_conf(); |
| if (jconf != null) { |
| conf.putAll((Map<String, Object>) JSONValue.parseWithException(jconf)); |
| } |
| return conf; |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| @SuppressWarnings("unchecked") |
| public static void validateBasic(StormTopology topology) throws InvalidTopologyException { |
| validateIds(topology); |
| |
| for (StormTopology._Fields field : Thrift.getSpoutFields()) { |
| Map<String, Object> spoutComponents = (Map<String, Object>) topology.getFieldValue(field); |
| if (spoutComponents != null) { |
| for (Object obj : spoutComponents.values()) { |
| ComponentCommon common = getComponentCommon(obj); |
| if (!isEmptyInputs(common)) { |
| throw new WrappedInvalidTopologyException("May not declare inputs for a spout"); |
| } |
| } |
| } |
| } |
| |
| Map<String, Object> componentMap = allComponents(topology); |
| for (Object componentObj : componentMap.values()) { |
| Map<String, Object> conf = componentConf(componentObj); |
| ComponentCommon common = getComponentCommon(componentObj); |
| int parallelismHintNum = Thrift.getParallelismHint(common); |
| Integer taskNum = ObjectReader.getInt(conf.get(Config.TOPOLOGY_TASKS), 0); |
| if (taskNum > 0 && parallelismHintNum <= 0) { |
| throw new WrappedInvalidTopologyException( |
| "Number of executors must be greater than 0 when number of tasks is greater than 0"); |
| } |
| } |
| } |
| |
| private static Set<String> getStreamOutputFields(Map<String, StreamInfo> streams) { |
| Set<String> outputFields = new HashSet<>(); |
| for (StreamInfo streamInfo : streams.values()) { |
| outputFields.addAll(streamInfo.get_output_fields()); |
| } |
| return outputFields; |
| } |
| |
| public static void validateStructure(StormTopology topology) throws InvalidTopologyException { |
| Map<String, Object> componentMap = allComponents(topology); |
| for (Map.Entry<String, Object> entry : componentMap.entrySet()) { |
| String componentId = entry.getKey(); |
| ComponentCommon common = getComponentCommon(entry.getValue()); |
| Map<GlobalStreamId, Grouping> inputs = common.get_inputs(); |
| for (Map.Entry<GlobalStreamId, Grouping> input : inputs.entrySet()) { |
| String sourceStreamId = input.getKey().get_streamId(); |
| String sourceComponentId = input.getKey().get_componentId(); |
| if (!componentMap.keySet().contains(sourceComponentId)) { |
| throw new WrappedInvalidTopologyException("Component: [" + componentId |
| + "] subscribes from non-existent component [" + sourceComponentId + "]"); |
| } |
| |
| ComponentCommon sourceComponent = getComponentCommon(componentMap.get(sourceComponentId)); |
| if (!sourceComponent.get_streams().containsKey(sourceStreamId)) { |
| throw new WrappedInvalidTopologyException("Component: [" + componentId |
| + "] subscribes from non-existent stream: " |
| + "[" + sourceStreamId + "] of component [" + sourceComponentId + "]"); |
| } |
| |
| Grouping grouping = input.getValue(); |
| if (Thrift.groupingType(grouping) == Grouping._Fields.FIELDS) { |
| List<String> fields = new ArrayList<>(grouping.get_fields()); |
| Map<String, StreamInfo> streams = sourceComponent.get_streams(); |
| Set<String> sourceOutputFields = getStreamOutputFields(streams); |
| fields.removeAll(sourceOutputFields); |
| if (fields.size() != 0) { |
| throw new WrappedInvalidTopologyException("Component: [" + componentId |
| + "] subscribes from stream: [" + sourceStreamId + "] of component " |
| + "[" + sourceComponentId + "] + with non-existent fields: " + fields); |
| } |
| } |
| } |
| } |
| } |
| |
| public static Map<GlobalStreamId, Grouping> ackerInputs(StormTopology topology) { |
| Map<GlobalStreamId, Grouping> inputs = new HashMap<>(); |
| Set<String> boltIds = topology.get_bolts().keySet(); |
| Set<String> spoutIds = topology.get_spouts().keySet(); |
| |
| for (String id : spoutIds) { |
| inputs.put(Utils.getGlobalStreamId(id, Acker.ACKER_INIT_STREAM_ID), |
| Thrift.prepareFieldsGrouping(Arrays.asList("id"))); |
| } |
| |
| for (String id : boltIds) { |
| inputs.put(Utils.getGlobalStreamId(id, Acker.ACKER_ACK_STREAM_ID), |
| Thrift.prepareFieldsGrouping(Arrays.asList("id"))); |
| inputs.put(Utils.getGlobalStreamId(id, Acker.ACKER_FAIL_STREAM_ID), |
| Thrift.prepareFieldsGrouping(Arrays.asList("id"))); |
| inputs.put(Utils.getGlobalStreamId(id, Acker.ACKER_RESET_TIMEOUT_STREAM_ID), |
| Thrift.prepareFieldsGrouping(Arrays.asList("id"))); |
| } |
| return inputs; |
| } |
| |
| public static IBolt makeAckerBolt() { |
| return _instance.makeAckerBoltImpl(); |
| } |
| |
| @SuppressWarnings("unchecked") |
| public static void addAcker(Map<String, Object> conf, StormTopology topology) { |
| |
| Map<String, StreamInfo> outputStreams = new HashMap<String, StreamInfo>(); |
| outputStreams.put(Acker.ACKER_ACK_STREAM_ID, Thrift.directOutputFields(Arrays.asList("id", "time-delta-ms"))); |
| outputStreams.put(Acker.ACKER_FAIL_STREAM_ID, Thrift.directOutputFields(Arrays.asList("id", "time-delta-ms"))); |
| outputStreams.put(Acker.ACKER_RESET_TIMEOUT_STREAM_ID, Thrift.directOutputFields(Arrays.asList("id", "time-delta-ms"))); |
| |
| Map<String, Object> ackerConf = new HashMap<>(); |
| int ackerNum = |
| ObjectReader.getInt(conf.get(Config.TOPOLOGY_ACKER_EXECUTORS), ObjectReader.getInt(conf.get(Config.TOPOLOGY_WORKERS))); |
| ackerConf.put(Config.TOPOLOGY_TASKS, ackerNum); |
| ackerConf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS))); |
| |
| Map<GlobalStreamId, Grouping> inputs = ackerInputs(topology); |
| Bolt acker = Thrift.prepareSerializedBoltDetails(inputs, makeAckerBolt(), outputStreams, ackerNum, ackerConf); |
| |
| for (Bolt bolt : topology.get_bolts().values()) { |
| ComponentCommon common = bolt.get_common(); |
| common.put_to_streams(Acker.ACKER_ACK_STREAM_ID, Thrift.outputFields(Arrays.asList("id", "ack-val"))); |
| common.put_to_streams(Acker.ACKER_FAIL_STREAM_ID, Thrift.outputFields(Arrays.asList("id"))); |
| common.put_to_streams(Acker.ACKER_RESET_TIMEOUT_STREAM_ID, Thrift.outputFields(Arrays.asList("id"))); |
| } |
| |
| for (SpoutSpec spout : topology.get_spouts().values()) { |
| ComponentCommon common = spout.get_common(); |
| Map<String, Object> spoutConf = componentConf(spout); |
| spoutConf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, |
| ObjectReader.getInt(conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS))); |
| common.set_json_conf(JSONValue.toJSONString(spoutConf)); |
| common.put_to_streams(Acker.ACKER_INIT_STREAM_ID, |
| Thrift.outputFields(Arrays.asList("id", "init-val", "spout-task"))); |
| common.put_to_inputs(Utils.getGlobalStreamId(Acker.ACKER_COMPONENT_ID, Acker.ACKER_ACK_STREAM_ID), |
| Thrift.prepareDirectGrouping()); |
| common.put_to_inputs(Utils.getGlobalStreamId(Acker.ACKER_COMPONENT_ID, Acker.ACKER_FAIL_STREAM_ID), |
| Thrift.prepareDirectGrouping()); |
| common.put_to_inputs(Utils.getGlobalStreamId(Acker.ACKER_COMPONENT_ID, Acker.ACKER_RESET_TIMEOUT_STREAM_ID), |
| Thrift.prepareDirectGrouping()); |
| } |
| |
| topology.put_to_bolts(Acker.ACKER_COMPONENT_ID, acker); |
| } |
| |
| public static ComponentCommon getComponentCommon(Object component) { |
| ComponentCommon common = null; |
| if (component instanceof StateSpoutSpec) { |
| common = ((StateSpoutSpec) component).get_common(); |
| } else if (component instanceof SpoutSpec) { |
| common = ((SpoutSpec) component).get_common(); |
| } else if (component instanceof Bolt) { |
| common = ((Bolt) component).get_common(); |
| } |
| return common; |
| } |
| |
| public static void addMetricStreams(StormTopology topology) { |
| for (Object component : allComponents(topology).values()) { |
| ComponentCommon common = getComponentCommon(component); |
| StreamInfo streamInfo = Thrift.outputFields(Arrays.asList("task-info", "data-points")); |
| common.put_to_streams(Constants.METRICS_STREAM_ID, streamInfo); |
| } |
| } |
| |
| public static void addSystemStreams(StormTopology topology) { |
| for (Object component : allComponents(topology).values()) { |
| ComponentCommon common = getComponentCommon(component); |
| StreamInfo streamInfo = Thrift.outputFields(Arrays.asList("event")); |
| common.put_to_streams(SYSTEM_STREAM_ID, streamInfo); |
| } |
| } |
| |
| public static List<String> eventLoggerBoltFields() { |
| return Arrays.asList(EventLoggerBolt.FIELD_COMPONENT_ID, EventLoggerBolt.FIELD_MESSAGE_ID, |
| EventLoggerBolt.FIELD_TS, EventLoggerBolt.FIELD_VALUES); |
| } |
| |
| public static Map<GlobalStreamId, Grouping> eventLoggerInputs(StormTopology topology) { |
| Map<GlobalStreamId, Grouping> inputs = new HashMap<GlobalStreamId, Grouping>(); |
| Set<String> allIds = new HashSet<String>(); |
| allIds.addAll(topology.get_bolts().keySet()); |
| allIds.addAll(topology.get_spouts().keySet()); |
| |
| for (String id : allIds) { |
| inputs.put(Utils.getGlobalStreamId(id, EVENTLOGGER_STREAM_ID), |
| Thrift.prepareFieldsGrouping(Arrays.asList("component-id"))); |
| } |
| return inputs; |
| } |
| |
| public static void addEventLogger(Map<String, Object> conf, StormTopology topology) { |
| Integer numExecutors = ObjectReader.getInt(conf.get(Config.TOPOLOGY_EVENTLOGGER_EXECUTORS), |
| ObjectReader.getInt(conf.get(Config.TOPOLOGY_WORKERS))); |
| if (numExecutors == null || numExecutors == 0) { |
| return; |
| } |
| HashMap<String, Object> componentConf = new HashMap<>(); |
| componentConf.put(Config.TOPOLOGY_TASKS, numExecutors); |
| componentConf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, ObjectReader.getInt(conf.get(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS))); |
| Bolt eventLoggerBolt = Thrift.prepareSerializedBoltDetails( |
| eventLoggerInputs(topology), new EventLoggerBolt(), null, numExecutors, componentConf); |
| |
| for (Object component : allComponents(topology).values()) { |
| ComponentCommon common = getComponentCommon(component); |
| common.put_to_streams(EVENTLOGGER_STREAM_ID, Thrift.outputFields(eventLoggerBoltFields())); |
| } |
| topology.put_to_bolts(EVENTLOGGER_COMPONENT_ID, eventLoggerBolt); |
| } |
| |
| @SuppressWarnings("unchecked") |
| public static Map<String, Bolt> metricsConsumerBoltSpecs(Map<String, Object> conf, StormTopology topology) { |
| Map<String, Bolt> metricsConsumerBolts = new HashMap<>(); |
| |
| Set<String> componentIdsEmitMetrics = new HashSet<>(); |
| componentIdsEmitMetrics.addAll(allComponents(topology).keySet()); |
| componentIdsEmitMetrics.add(Constants.SYSTEM_COMPONENT_ID); |
| |
| Map<GlobalStreamId, Grouping> inputs = new HashMap<>(); |
| for (String componentId : componentIdsEmitMetrics) { |
| inputs.put(Utils.getGlobalStreamId(componentId, Constants.METRICS_STREAM_ID), Thrift.prepareShuffleGrouping()); |
| } |
| |
| List<Map<String, Object>> registerInfo = (List<Map<String, Object>>) conf.get(Config.TOPOLOGY_METRICS_CONSUMER_REGISTER); |
| if (registerInfo != null) { |
| Map<String, Integer> classOccurrencesMap = new HashMap<String, Integer>(); |
| for (Map<String, Object> info : registerInfo) { |
| String className = (String) info.get(TOPOLOGY_METRICS_CONSUMER_CLASS); |
| Object argument = info.get(TOPOLOGY_METRICS_CONSUMER_ARGUMENT); |
| Integer maxRetainMetricTuples = ObjectReader.getInt(info.get( |
| TOPOLOGY_METRICS_CONSUMER_MAX_RETAIN_METRIC_TUPLES), 100); |
| Integer phintNum = ObjectReader.getInt(info.get(TOPOLOGY_METRICS_CONSUMER_PARALLELISM_HINT), 1); |
| Map<String, Object> metricsConsumerConf = new HashMap<String, Object>(); |
| metricsConsumerConf.put(Config.TOPOLOGY_TASKS, phintNum); |
| List<String> whitelist = (List<String>) info.get( |
| TOPOLOGY_METRICS_CONSUMER_WHITELIST); |
| List<String> blacklist = (List<String>) info.get( |
| TOPOLOGY_METRICS_CONSUMER_BLACKLIST); |
| FilterByMetricName filterPredicate = new FilterByMetricName(whitelist, blacklist); |
| Boolean expandMapType = ObjectReader.getBoolean(info.get( |
| TOPOLOGY_METRICS_CONSUMER_EXPAND_MAP_TYPE), false); |
| String metricNameSeparator = ObjectReader.getString(info.get( |
| TOPOLOGY_METRICS_CONSUMER_METRIC_NAME_SEPARATOR), "."); |
| DataPointExpander expander = new DataPointExpander(expandMapType, metricNameSeparator); |
| MetricsConsumerBolt boltInstance = new MetricsConsumerBolt(className, argument, |
| maxRetainMetricTuples, filterPredicate, expander); |
| Bolt metricsConsumerBolt = Thrift.prepareSerializedBoltDetails(inputs, |
| boltInstance, null, phintNum, metricsConsumerConf); |
| |
| String id = className; |
| if (classOccurrencesMap.containsKey(className)) { |
| // e.g. [\"a\", \"b\", \"a\"]) => [\"a\", \"b\", \"a#2\"]" |
| int occurrenceNum = classOccurrencesMap.get(className); |
| occurrenceNum++; |
| classOccurrencesMap.put(className, occurrenceNum); |
| id = Constants.METRICS_COMPONENT_ID_PREFIX + className + "#" + occurrenceNum; |
| } else { |
| id = Constants.METRICS_COMPONENT_ID_PREFIX + className; |
| classOccurrencesMap.put(className, 1); |
| } |
| metricsConsumerBolts.put(id, metricsConsumerBolt); |
| } |
| } |
| return metricsConsumerBolts; |
| } |
| |
| public static void addMetricComponents(Map<String, Object> conf, StormTopology topology) { |
| Map<String, Bolt> metricsConsumerBolts = metricsConsumerBoltSpecs(conf, topology); |
| for (Map.Entry<String, Bolt> entry : metricsConsumerBolts.entrySet()) { |
| topology.put_to_bolts(entry.getKey(), entry.getValue()); |
| } |
| } |
| |
| @SuppressWarnings("unused") |
| public static void addSystemComponents(Map<String, Object> conf, StormTopology topology) { |
| Map<String, StreamInfo> outputStreams = new HashMap<>(); |
| outputStreams.put(Constants.SYSTEM_TICK_STREAM_ID, Thrift.outputFields(Arrays.asList("rate_secs"))); |
| outputStreams.put(Constants.SYSTEM_FLUSH_STREAM_ID, Thrift.outputFields(Arrays.asList())); |
| outputStreams.put(Constants.METRICS_TICK_STREAM_ID, Thrift.outputFields(Arrays.asList("interval"))); |
| outputStreams.put(Constants.CREDENTIALS_CHANGED_STREAM_ID, Thrift.outputFields(Arrays.asList("creds"))); |
| |
| Map<String, Object> boltConf = new HashMap<>(); |
| boltConf.put(Config.TOPOLOGY_TASKS, 0); |
| |
| Bolt systemBoltSpec = Thrift.prepareSerializedBoltDetails(null, new SystemBolt(), outputStreams, 0, boltConf); |
| topology.put_to_bolts(Constants.SYSTEM_COMPONENT_ID, systemBoltSpec); |
| } |
| |
| public static StormTopology systemTopology(Map<String, Object> topoConf, StormTopology topology) throws InvalidTopologyException { |
| return _instance.systemTopologyImpl(topoConf, topology); |
| } |
| |
| public static boolean hasAckers(Map<String, Object> topoConf) { |
| Object ackerNum = topoConf.get(Config.TOPOLOGY_ACKER_EXECUTORS); |
| return ackerNum == null || ObjectReader.getInt(ackerNum) > 0; |
| } |
| |
| public static boolean hasEventLoggers(Map<String, Object> topoConf) { |
| Object eventLoggerNum = topoConf.get(Config.TOPOLOGY_EVENTLOGGER_EXECUTORS); |
| return eventLoggerNum == null || ObjectReader.getInt(eventLoggerNum) > 0; |
| } |
| |
| public static int numStartExecutors(Object component) throws InvalidTopologyException { |
| ComponentCommon common = getComponentCommon(component); |
| return Thrift.getParallelismHint(common); |
| } |
| |
| public static Map<Integer, String> stormTaskInfo(StormTopology userTopology, Map<String, Object> topoConf) throws |
| InvalidTopologyException { |
| return _instance.stormTaskInfoImpl(userTopology, topoConf); |
| } |
| |
| public static List<Integer> executorIdToTasks(List<Long> executorId) { |
| List<Integer> taskIds = new ArrayList<>(); |
| int taskId = executorId.get(0).intValue(); |
| while (taskId <= executorId.get(1).intValue()) { |
| taskIds.add(taskId); |
| taskId++; |
| } |
| return taskIds; |
| } |
| |
| public static Map<Integer, NodeInfo> taskToNodeport(Map<List<Long>, NodeInfo> executorToNodePort) { |
| Map<Integer, NodeInfo> tasksToNodePort = new HashMap<>(); |
| for (Map.Entry<List<Long>, NodeInfo> entry : executorToNodePort.entrySet()) { |
| List<Integer> taskIds = executorIdToTasks(entry.getKey()); |
| for (Integer taskId : taskIds) { |
| tasksToNodePort.put(taskId, entry.getValue()); |
| } |
| } |
| return tasksToNodePort; |
| } |
| |
| public static IAuthorizer mkAuthorizationHandler(String klassName, Map<String, Object> conf) |
| throws IllegalAccessException, InstantiationException, ClassNotFoundException { |
| return _instance.mkAuthorizationHandlerImpl(klassName, conf); |
| } |
| |
| @SuppressWarnings("unchecked") |
| public static WorkerTopologyContext makeWorkerContext(Map<String, Object> workerData) { |
| try { |
| StormTopology stormTopology = (StormTopology) workerData.get(Constants.SYSTEM_TOPOLOGY); |
| Map<String, Object> topoConf = (Map) workerData.get(Constants.STORM_CONF); |
| Map<Integer, String> taskToComponent = (Map<Integer, String>) workerData.get(Constants.TASK_TO_COMPONENT); |
| Map<String, List<Integer>> componentToSortedTasks = |
| (Map<String, List<Integer>>) workerData.get(Constants.COMPONENT_TO_SORTED_TASKS); |
| Map<String, Map<String, Fields>> componentToStreamToFields = |
| (Map<String, Map<String, Fields>>) workerData.get(Constants.COMPONENT_TO_STREAM_TO_FIELDS); |
| String stormId = (String) workerData.get(Constants.STORM_ID); |
| Map<String, Object> conf = (Map) workerData.get(Constants.CONF); |
| Integer port = (Integer) workerData.get(Constants.PORT); |
| String codeDir = ConfigUtils.supervisorStormResourcesPath(ConfigUtils.supervisorStormDistRoot(conf, stormId)); |
| String pidDir = ConfigUtils.workerPidsRoot(conf, stormId); |
| List<Integer> workerTasks = (List<Integer>) workerData.get(Constants.TASK_IDS); |
| Map<String, Object> defaultResources = (Map<String, Object>) workerData.get(Constants.DEFAULT_SHARED_RESOURCES); |
| Map<String, Object> userResources = (Map<String, Object>) workerData.get(Constants.USER_SHARED_RESOURCES); |
| return new WorkerTopologyContext(stormTopology, topoConf, taskToComponent, componentToSortedTasks, |
| componentToStreamToFields, stormId, codeDir, pidDir, port, workerTasks, defaultResources, |
| userResources); |
| } catch (IOException e) { |
| throw Utils.wrapInRuntime(e); |
| } |
| } |
| |
| public IBolt makeAckerBoltImpl() { |
| return new Acker(); |
| } |
| |
| protected StormTopology systemTopologyImpl(Map<String, Object> topoConf, StormTopology topology) throws InvalidTopologyException { |
| validateBasic(topology); |
| |
| StormTopology ret = topology.deepCopy(); |
| addAcker(topoConf, ret); |
| if (hasEventLoggers(topoConf)) { |
| addEventLogger(topoConf, ret); |
| } |
| addMetricComponents(topoConf, ret); |
| addSystemComponents(topoConf, ret); |
| addMetricStreams(ret); |
| addSystemStreams(ret); |
| |
| validateStructure(ret); |
| |
| return ret; |
| } |
| |
| /* |
| * Returns map from task -> componentId |
| */ |
| protected Map<Integer, String> stormTaskInfoImpl(StormTopology userTopology, Map<String, Object> topoConf) throws |
| InvalidTopologyException { |
| Map<Integer, String> taskIdToComponentId = new HashMap<>(); |
| |
| StormTopology systemTopology = systemTopology(topoConf, userTopology); |
| Map<String, Object> components = allComponents(systemTopology); |
| Map<String, Integer> componentIdToTaskNum = new TreeMap<>(); |
| for (Map.Entry<String, Object> entry : components.entrySet()) { |
| Map<String, Object> conf = componentConf(entry.getValue()); |
| Object taskNum = conf.get(Config.TOPOLOGY_TASKS); |
| componentIdToTaskNum.put(entry.getKey(), ObjectReader.getInt(taskNum)); |
| } |
| |
| int taskId = 1; |
| for (Map.Entry<String, Integer> entry : componentIdToTaskNum.entrySet()) { |
| String componentId = entry.getKey(); |
| Integer taskNum = entry.getValue(); |
| while (taskNum > 0) { |
| taskIdToComponentId.put(taskId, componentId); |
| taskNum--; |
| taskId++; |
| } |
| } |
| return taskIdToComponentId; |
| } |
| |
| protected IAuthorizer mkAuthorizationHandlerImpl(String klassName, Map<String, Object> conf) |
| throws ClassNotFoundException, IllegalAccessException, InstantiationException { |
| IAuthorizer aznHandler = null; |
| if (StringUtils.isNotBlank(klassName)) { |
| Class<?> aznClass = Class.forName(klassName); |
| if (aznClass != null) { |
| aznHandler = (IAuthorizer) aznClass.newInstance(); |
| if (aznHandler != null) { |
| aznHandler.prepare(conf); |
| } |
| LOG.debug("authorization class name:{}, class:{}, handler:{}", klassName, aznClass, aznHandler); |
| } |
| } |
| |
| return aznHandler; |
| } |
| } |
