| /* |
| * 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; |
| |
| import java.lang.reflect.Method; |
| import java.nio.ByteBuffer; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ThreadLocalRandom; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.function.UnaryOperator; |
| import org.apache.storm.blobstore.BlobStore; |
| import org.apache.storm.cluster.ClusterStateContext; |
| import org.apache.storm.cluster.ClusterUtils; |
| import org.apache.storm.cluster.DaemonType; |
| import org.apache.storm.cluster.IStateStorage; |
| import org.apache.storm.cluster.IStormClusterState; |
| import org.apache.storm.daemon.Acker; |
| import org.apache.storm.daemon.DaemonCommon; |
| import org.apache.storm.daemon.Shutdownable; |
| import org.apache.storm.daemon.StormCommon; |
| import org.apache.storm.daemon.nimbus.Nimbus; |
| import org.apache.storm.daemon.nimbus.Nimbus.StandaloneINimbus; |
| import org.apache.storm.daemon.nimbus.TopoCache; |
| import org.apache.storm.daemon.supervisor.ReadClusterState; |
| import org.apache.storm.daemon.supervisor.StandaloneSupervisor; |
| import org.apache.storm.daemon.supervisor.Supervisor; |
| import org.apache.storm.executor.LocalExecutor; |
| import org.apache.storm.generated.AlreadyAliveException; |
| import org.apache.storm.generated.AuthorizationException; |
| import org.apache.storm.generated.BeginDownloadResult; |
| import org.apache.storm.generated.ClusterSummary; |
| import org.apache.storm.generated.ComponentPageInfo; |
| import org.apache.storm.generated.Credentials; |
| import org.apache.storm.generated.GetInfoOptions; |
| import org.apache.storm.generated.InvalidTopologyException; |
| import org.apache.storm.generated.KeyAlreadyExistsException; |
| import org.apache.storm.generated.KeyNotFoundException; |
| import org.apache.storm.generated.KillOptions; |
| import org.apache.storm.generated.ListBlobsResult; |
| import org.apache.storm.generated.LogConfig; |
| import org.apache.storm.generated.Nimbus.Iface; |
| import org.apache.storm.generated.Nimbus.Processor; |
| import org.apache.storm.generated.NimbusSummary; |
| import org.apache.storm.generated.NotAliveException; |
| import org.apache.storm.generated.OwnerResourceSummary; |
| import org.apache.storm.generated.ProfileAction; |
| import org.apache.storm.generated.ProfileRequest; |
| import org.apache.storm.generated.ReadableBlobMeta; |
| import org.apache.storm.generated.RebalanceOptions; |
| import org.apache.storm.generated.SettableBlobMeta; |
| import org.apache.storm.generated.StormTopology; |
| import org.apache.storm.generated.SubmitOptions; |
| import org.apache.storm.generated.SupervisorAssignments; |
| import org.apache.storm.generated.SupervisorPageInfo; |
| import org.apache.storm.generated.SupervisorWorkerHeartbeat; |
| import org.apache.storm.generated.SupervisorWorkerHeartbeats; |
| import org.apache.storm.generated.TopologyHistoryInfo; |
| import org.apache.storm.generated.TopologyInfo; |
| import org.apache.storm.generated.TopologyPageInfo; |
| import org.apache.storm.generated.WorkerMetrics; |
| import org.apache.storm.messaging.IContext; |
| import org.apache.storm.messaging.local.Context; |
| import org.apache.storm.metric.StormMetricsRegistry; |
| import org.apache.storm.nimbus.ILeaderElector; |
| import org.apache.storm.scheduler.INimbus; |
| import org.apache.storm.scheduler.ISupervisor; |
| import org.apache.storm.security.auth.IGroupMappingServiceProvider; |
| import org.apache.storm.security.auth.ThriftConnectionType; |
| import org.apache.storm.security.auth.ThriftServer; |
| import org.apache.storm.shade.org.apache.zookeeper.server.ServerConfig; |
| import org.apache.storm.task.IBolt; |
| import org.apache.storm.testing.InProcessZookeeper; |
| import org.apache.storm.testing.NonRichBoltTracker; |
| import org.apache.storm.testing.TmpPath; |
| import org.apache.storm.testing.TrackedTopology; |
| import org.apache.storm.thrift.TException; |
| import org.apache.storm.utils.ConfigUtils; |
| import org.apache.storm.utils.DRPCClient; |
| import org.apache.storm.utils.NimbusClient; |
| import org.apache.storm.utils.ObjectReader; |
| import org.apache.storm.utils.RegisteredGlobalState; |
| import org.apache.storm.utils.StormCommonInstaller; |
| import org.apache.storm.utils.Time; |
| import org.apache.storm.utils.Time.SimulatedTime; |
| import org.apache.storm.utils.Utils; |
| import org.apache.storm.utils.WrappedAuthorizationException; |
| import org.apache.storm.utils.WrappedKeyNotFoundException; |
| import org.json.simple.JSONValue; |
| import org.json.simple.parser.ParseException; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * A stand alone storm cluster that runs inside a single process. It is intended to be used for testing. Both internal testing for Apache |
| * Storm itself and for people building storm topologies. |
| * <p> |
| * LocalCluster is an AutoCloseable so if you are using it in tests you can use a try block to be sure it is shut down. |
| * </p> |
| * try (LocalCluster cluster = new LocalCluster()) { // Do some tests } // The cluster has been shut down. |
| */ |
| public class LocalCluster implements ILocalClusterTrackedTopologyAware, Iface { |
| public static final KillOptions KILL_NOW = new KillOptions(); |
| private static final Logger LOG = LoggerFactory.getLogger(LocalCluster.class); |
| private static final long DEFAULT_ZK_PORT = 2181; |
| |
| static { |
| KILL_NOW.set_wait_secs(0); |
| } |
| |
| private final Nimbus nimbus; |
| //This is very private and does not need to be exposed |
| private int portCounter; |
| private final Map<String, Object> daemonConf; |
| private final List<Supervisor> supervisors; |
| private final IStateStorage state; |
| private final IStormClusterState clusterState; |
| private final String stormHomeBackup; |
| private final List<TmpPath> tmpDirs; |
| private final InProcessZookeeper zookeeper; |
| private final IContext sharedContext; |
| private final ThriftServer thriftServer; |
| private final String trackId; |
| private final StormCommonInstaller commonInstaller; |
| private final SimulatedTime time; |
| private final NimbusClient.LocalOverride nimbusOverride; |
| private final StormMetricsRegistry metricRegistry; |
| |
| /** |
| * Create a default LocalCluster. |
| * |
| * @throws Exception on any error |
| */ |
| public LocalCluster() throws Exception { |
| this(new Builder().withDaemonConf(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, true)); |
| } |
| |
| /** |
| * Create a LocalCluster that connects to an existing Zookeeper instance. |
| * |
| * @param zkHost the host for ZK |
| * @param zkPort the port for ZK |
| * @throws Exception on any error |
| */ |
| public LocalCluster(String zkHost, Long zkPort) throws Exception { |
| this(new Builder().withDaemonConf(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, true) |
| .withDaemonConf(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList(zkHost)) |
| .withDaemonConf(Config.STORM_ZOOKEEPER_PORT, zkPort)); |
| } |
| |
| @SuppressWarnings("deprecation") |
| private LocalCluster(Builder builder) throws Exception { |
| if (builder.simulateTime) { |
| time = new SimulatedTime(); |
| } else { |
| time = null; |
| } |
| boolean success = false; |
| try { |
| this.trackId = builder.trackId; |
| if (trackId != null) { |
| ConcurrentHashMap<String, AtomicInteger> metrics = new ConcurrentHashMap<>(); |
| metrics.put("spout-emitted", new AtomicInteger(0)); |
| metrics.put("transferred", new AtomicInteger(0)); |
| metrics.put("processed", new AtomicInteger(0)); |
| this.commonInstaller = new StormCommonInstaller(new TrackedStormCommon(this.trackId)); |
| LOG.warn("Adding tracked metrics for ID {}", this.trackId); |
| RegisteredGlobalState.setState(this.trackId, metrics); |
| LocalExecutor.setTrackId(this.trackId); |
| } else { |
| this.commonInstaller = null; |
| } |
| |
| this.tmpDirs = new ArrayList<>(); |
| this.supervisors = new ArrayList<>(); |
| TmpPath nimbusTmp = new TmpPath(); |
| this.tmpDirs.add(nimbusTmp); |
| stormHomeBackup = System.getProperty(ConfigUtils.STORM_HOME); |
| TmpPath stormHome = new TmpPath(); |
| if (!stormHome.getFile().mkdirs()) { |
| throw new IllegalStateException("Failed to create storm.home directory " + stormHome.getPath()); |
| } |
| this.tmpDirs.add(stormHome); |
| System.setProperty(ConfigUtils.STORM_HOME, stormHome.getPath()); |
| Map<String, Object> conf = ConfigUtils.readStormConfig(); |
| conf.put(Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS, true); |
| conf.put(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, false); |
| conf.put(Config.TOPOLOGY_TRIDENT_BATCH_EMIT_INTERVAL_MILLIS, 50); |
| conf.put(Config.STORM_CLUSTER_MODE, "local"); |
| conf.put(Config.BLOBSTORE_DIR, nimbusTmp.getPath()); |
| conf.put(Config.TOPOLOGY_MIN_REPLICATION_COUNT, 1); |
| |
| InProcessZookeeper zookeeper = null; |
| if (!builder.daemonConf.containsKey(Config.STORM_ZOOKEEPER_SERVERS)) { |
| zookeeper = new InProcessZookeeper(); |
| conf.put(Config.STORM_ZOOKEEPER_PORT, zookeeper.getPort()); |
| conf.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList("localhost")); |
| } |
| this.zookeeper = zookeeper; |
| conf.putAll(builder.daemonConf); |
| this.daemonConf = new HashMap<>(conf); |
| this.metricRegistry = new StormMetricsRegistry(); |
| |
| this.portCounter = builder.supervisorSlotPortMin; |
| ClusterStateContext cs = new ClusterStateContext(DaemonType.NIMBUS, daemonConf); |
| this.state = ClusterUtils.mkStateStorage(this.daemonConf, null, cs); |
| if (builder.clusterState == null) { |
| clusterState = ClusterUtils.mkStormClusterState(this.daemonConf, cs); |
| } else { |
| this.clusterState = builder.clusterState; |
| } |
| if (!Time.isSimulating()) { |
| //Ensure Nimbus assigns topologies as quickly as possible |
| conf.put(DaemonConfig.NIMBUS_MONITOR_FREQ_SECS, 1); |
| } |
| //Set it for nimbus only |
| conf.put(Config.STORM_LOCAL_DIR, nimbusTmp.getPath()); |
| Nimbus nimbus = new Nimbus(conf, builder.inimbus == null ? new StandaloneINimbus() : builder.inimbus, |
| this.getClusterState(), null, builder.store, builder.topoCache, builder.leaderElector, |
| builder.groupMapper, metricRegistry); |
| if (builder.nimbusWrapper != null) { |
| nimbus = builder.nimbusWrapper.apply(nimbus); |
| } |
| this.nimbus = nimbus; |
| this.nimbus.launchServer(); |
| if (!this.nimbus.awaitLeadership(Testing.TEST_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { |
| //Ensure Nimbus has leadership, otherwise topology submission will fail. |
| throw new RuntimeException("LocalCluster Nimbus failed to gain leadership."); |
| } |
| IContext context = null; |
| if (!ObjectReader.getBoolean(this.daemonConf.get(Config.STORM_LOCAL_MODE_ZMQ), false)) { |
| context = new Context(); |
| context.prepare(this.daemonConf); |
| } |
| this.sharedContext = context; |
| this.thriftServer = builder.nimbusDaemon ? startNimbusDaemon(this.daemonConf, this.nimbus) : null; |
| |
| for (int i = 0; i < builder.supervisors; i++) { |
| addSupervisor(builder.portsPerSupervisor, null, null); |
| } |
| |
| //Wait for a leader to be elected (or topology submission can be rejected) |
| try { |
| long timeoutAfter = System.currentTimeMillis() + 10_000; |
| while (!hasLeader()) { |
| if (timeoutAfter > System.currentTimeMillis()) { |
| throw new IllegalStateException("Timed out waiting for nimbus to become the leader"); |
| } |
| Thread.sleep(1); |
| } |
| } catch (Exception e) { |
| //Ignore any exceptions we might be doing a test for authentication |
| } |
| if (thriftServer == null) { |
| //We don't want to override the client if there is a thrift server up and running, or we would not test any |
| // Of the actual thrift code |
| this.nimbusOverride = new NimbusClient.LocalOverride(this); |
| } else { |
| this.nimbusOverride = null; |
| } |
| success = true; |
| |
| metricRegistry.startMetricsReporters(daemonConf); |
| } finally { |
| if (!success) { |
| close(); |
| } |
| } |
| } |
| |
| private static ThriftServer startNimbusDaemon(Map<String, Object> conf, Nimbus nimbus) { |
| ThriftServer ret = new ThriftServer(conf, new Processor<>(nimbus), ThriftConnectionType.NIMBUS); |
| LOG.info("Starting Nimbus server..."); |
| new Thread(() -> ret.serve()).start(); |
| return ret; |
| } |
| |
| private static boolean areAllWorkersWaiting() { |
| boolean ret = true; |
| for (Shutdownable s : ProcessSimulator.getAllProcessHandles()) { |
| if (s instanceof DaemonCommon) { |
| ret = ret && ((DaemonCommon) s).isWaiting(); |
| } |
| } |
| return ret; |
| } |
| |
| /** |
| * Run c with a local mode cluster overriding the NimbusClient and DRPCClient calls. NOTE local mode override happens by default now |
| * unless netty is turned on for the local cluster. |
| * |
| * @param c the callable to run in this mode |
| * @param ttlSec the number of seconds to let the cluster run after c has completed |
| * @return the result of calling C |
| * |
| * @throws Exception on any Exception. |
| */ |
| public static <T> T withLocalModeOverride(Callable<T> c, long ttlSec) throws Exception { |
| return withLocalModeOverride(c, ttlSec, null); |
| } |
| |
| /** |
| * Run c with a local mode cluster overriding the NimbusClient and DRPCClient calls. NOTE local mode override happens by default now |
| * unless netty is turned on for the local cluster. |
| * |
| * @param c the callable to run in this mode |
| * @param ttlSec the number of seconds to let the cluster run after c has completed |
| * @param daemonConf configs to set for the daemon processes. |
| * @return the result of calling C |
| * |
| * @throws Exception on any Exception. |
| */ |
| @SuppressWarnings("checkstyle:VariableDeclarationUsageDistance") |
| public static <T> T withLocalModeOverride(Callable<T> c, long ttlSec, Map<String, Object> daemonConf) throws Exception { |
| LOG.info("\n\n\t\tSTARTING LOCAL MODE CLUSTER\n\n"); |
| Builder builder = new Builder(); |
| if (daemonConf != null) { |
| builder.withDaemonConf(daemonConf); |
| } |
| try (LocalCluster local = builder.build(); |
| LocalDRPC drpc = new LocalDRPC(local.metricRegistry); |
| DRPCClient.LocalOverride drpcOverride = new DRPCClient.LocalOverride(drpc)) { |
| |
| T ret = c.call(); |
| LOG.info("\n\n\t\tRUNNING LOCAL CLUSTER for {} seconds.\n\n", ttlSec); |
| Thread.sleep(ttlSec * 1000); |
| |
| LOG.info("\n\n\t\tSTOPPING LOCAL MODE CLUSTER\n\n"); |
| return ret; |
| } |
| } |
| |
| /** |
| * Main entry point to running in local mode. |
| * @param args arguments to be run in local mode |
| * @throws Exception on any error when running. |
| */ |
| public static void main(final String[] args) throws Exception { |
| if (args.length < 1) { |
| throw new IllegalArgumentException("No class was specified to run"); |
| } |
| |
| long ttl = 20; |
| String ttlString = System.getProperty("storm.local.sleeptime", "20"); |
| try { |
| ttl = Long.valueOf(ttlString); |
| } catch (NumberFormatException e) { |
| LOG.warn("could not parse the sleep time defaulting to {} seconds", ttl); |
| } |
| |
| Map<String, Object> daemonConf = new HashMap<>(); |
| String zkOverride = System.getProperty("storm.local.zookeeper"); |
| if (zkOverride != null) { |
| LOG.info("Using ZooKeeper at '{}' instead of in-process one.", zkOverride); |
| long zkPort = DEFAULT_ZK_PORT; |
| String zkHost = null; |
| if (zkOverride.contains(":")) { |
| String[] hostPort = zkOverride.split(":"); |
| zkHost = hostPort[0]; |
| zkPort = hostPort.length > 1 ? Long.parseLong(hostPort[1]) : DEFAULT_ZK_PORT; |
| } else { |
| zkHost = zkOverride; |
| } |
| daemonConf.put(Config.TOPOLOGY_ENABLE_MESSAGE_TIMEOUTS, true); |
| daemonConf.put(Config.STORM_ZOOKEEPER_SERVERS, Arrays.asList(zkHost)); |
| daemonConf.put(Config.STORM_ZOOKEEPER_PORT, zkPort); |
| } |
| |
| withLocalModeOverride(() -> { |
| String klass = args[0]; |
| String[] newArgs = Arrays.copyOfRange(args, 1, args.length); |
| Class<?> c = Class.forName(klass); |
| Method main = c.getDeclaredMethod("main", String[].class); |
| |
| LOG.info("\n\n\t\tRUNNING {} with args {}\n\n", main, Arrays.toString(newArgs)); |
| main.invoke(null, (Object) newArgs); |
| return (Void) null; |
| }, ttl, daemonConf); |
| |
| //Sometimes external things used with testing don't shut down all the way |
| System.exit(0); |
| } |
| |
| /** |
| * Checks if Nimbuses have elected a leader. |
| * |
| * @return true if there is a leader else false. |
| */ |
| private boolean hasLeader() throws AuthorizationException, TException { |
| ClusterSummary summary = getNimbus().getClusterInfo(); |
| if (summary.is_set_nimbuses()) { |
| for (NimbusSummary sum : summary.get_nimbuses()) { |
| if (sum.is_isLeader()) { |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Reference to nimbus. |
| * @return Nimbus itself so you can interact with it directly, if needed. |
| */ |
| public Nimbus getNimbus() { |
| return nimbus; |
| } |
| |
| /** |
| * Reference to metrics registry. |
| * @return The metrics registry for the local cluster. |
| */ |
| public StormMetricsRegistry getMetricRegistry() { |
| return metricRegistry; |
| } |
| |
| /** |
| * Get daemon configuration. |
| * @return the base config for the daemons |
| */ |
| public Map<String, Object> getDaemonConf() { |
| return new HashMap<>(daemonConf); |
| } |
| |
| @Override |
| public LocalTopology submitTopology(String topologyName, Map<String, Object> conf, StormTopology topology) |
| throws TException { |
| if (!Utils.isValidConf(conf)) { |
| throw new IllegalArgumentException("Topology conf is not json-serializable"); |
| } |
| getNimbus().submitTopology(topologyName, null, JSONValue.toJSONString(conf), Utils.addVersions(topology)); |
| |
| ISubmitterHook hook = (ISubmitterHook) Utils.getConfiguredClass(conf, Config.STORM_TOPOLOGY_SUBMISSION_NOTIFIER_PLUGIN); |
| if (hook != null) { |
| TopologyInfo topologyInfo = Utils.getTopologyInfo(topologyName, null, conf); |
| try { |
| hook.notify(topologyInfo, conf, topology); |
| } catch (IllegalAccessException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| return new LocalTopology(topologyName, topology); |
| } |
| |
| @Override |
| public LocalTopology submitTopology(String topologyName, Map<String, Object> conf, TrackedTopology topology) |
| throws TException { |
| return submitTopology(topologyName, conf, topology.getTopology()); |
| } |
| |
| @Override |
| public void submitTopology(String name, String uploadedJarLocation, String jsonConf, StormTopology topology) |
| throws AlreadyAliveException, InvalidTopologyException, AuthorizationException, TException { |
| try { |
| @SuppressWarnings("unchecked") |
| Map<String, Object> conf = (Map<String, Object>) JSONValue.parseWithException(jsonConf); |
| submitTopology(name, conf, topology); |
| } catch (ParseException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| @Override |
| public LocalTopology submitTopologyWithOpts(String topologyName, Map<String, Object> conf, StormTopology topology, |
| SubmitOptions submitOpts) |
| throws TException { |
| if (!Utils.isValidConf(conf)) { |
| throw new IllegalArgumentException("Topology conf is not json-serializable"); |
| } |
| getNimbus().submitTopologyWithOpts(topologyName, null, JSONValue.toJSONString(conf), Utils.addVersions(topology), submitOpts); |
| return new LocalTopology(topologyName, topology); |
| } |
| |
| @Override |
| public LocalTopology submitTopologyWithOpts(String topologyName, Map<String, Object> conf, TrackedTopology topology, |
| SubmitOptions submitOpts) |
| throws TException { |
| return submitTopologyWithOpts(topologyName, conf, topology.getTopology(), submitOpts); |
| } |
| |
| @Override |
| public void submitTopologyWithOpts(String name, String uploadedJarLocation, String jsonConf, StormTopology topology, |
| SubmitOptions options) |
| throws AlreadyAliveException, InvalidTopologyException, AuthorizationException, TException { |
| try { |
| @SuppressWarnings("unchecked") |
| Map<String, Object> conf = (Map<String, Object>) JSONValue.parseWithException(jsonConf); |
| submitTopologyWithOpts(name, conf, topology, options); |
| } catch (ParseException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| @Override |
| public void uploadNewCredentials(String topologyName, Credentials creds) throws TException { |
| getNimbus().uploadNewCredentials(topologyName, creds); |
| } |
| |
| @Override |
| public void killTopology(String topologyName) throws TException { |
| getNimbus().killTopology(topologyName); |
| } |
| |
| @Override |
| public void killTopologyWithOpts(String name, KillOptions options) throws TException { |
| getNimbus().killTopologyWithOpts(name, options); |
| } |
| |
| |
| @Override |
| public void activate(String topologyName) throws TException { |
| getNimbus().activate(topologyName); |
| } |
| |
| |
| @Override |
| public void deactivate(String topologyName) throws TException { |
| getNimbus().deactivate(topologyName); |
| } |
| |
| |
| @Override |
| public void rebalance(String name, RebalanceOptions options) throws TException { |
| getNimbus().rebalance(name, options); |
| } |
| |
| @Override |
| public void shutdown() { |
| try { |
| close(); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| @Override |
| public String getTopologyConf(String id) throws TException { |
| return getNimbus().getTopologyConf(id); |
| } |
| |
| @Override |
| public StormTopology getTopology(String id) throws TException { |
| return getNimbus().getTopology(id); |
| } |
| |
| @Override |
| public ClusterSummary getClusterInfo() throws TException { |
| return getNimbus().getClusterInfo(); |
| } |
| |
| @Override |
| public TopologyInfo getTopologyInfo(String id) throws TException { |
| return getNimbus().getTopologyInfo(id); |
| } |
| |
| public int getThriftServerPort() { |
| return thriftServer.getPort(); |
| } |
| |
| @Override |
| public boolean isRemoteBlobExists(String blobKey) throws AuthorizationException, TException { |
| throw new KeyNotFoundException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public synchronized void close() throws Exception { |
| if (metricRegistry != null) { |
| metricRegistry.stopMetricsReporters(); |
| } |
| if (nimbusOverride != null) { |
| nimbusOverride.close(); |
| } |
| if (nimbus != null) { |
| nimbus.shutdown(); |
| } |
| if (thriftServer != null) { |
| LOG.info("shutting down thrift server"); |
| try { |
| thriftServer.stop(); |
| } catch (Exception e) { |
| LOG.info("failed to stop thrift", e); |
| } |
| } |
| if (state != null) { |
| state.close(); |
| } |
| if (getClusterState() != null) { |
| getClusterState().disconnect(); |
| } |
| for (Supervisor s : supervisors) { |
| s.shutdownAllWorkers(null, ReadClusterState.THREAD_DUMP_ON_ERROR); |
| s.close(); |
| } |
| ProcessSimulator.killAllProcesses(); |
| if (zookeeper != null) { |
| LOG.info("Shutting down in process zookeeper"); |
| zookeeper.close(); |
| LOG.info("Done shutting down in process zookeeper"); |
| } |
| |
| for (TmpPath p : tmpDirs) { |
| p.close(); |
| } |
| |
| if (this.trackId != null) { |
| LOG.warn("Clearing tracked metrics for ID {}", this.trackId); |
| LocalExecutor.clearTrackId(); |
| RegisteredGlobalState.clearState(this.trackId); |
| } |
| |
| if (this.commonInstaller != null) { |
| this.commonInstaller.close(); |
| } |
| |
| if (time != null) { |
| time.close(); |
| } |
| |
| if (stormHomeBackup != null) { |
| System.setProperty(ConfigUtils.STORM_HOME, stormHomeBackup); |
| } else { |
| System.clearProperty(ConfigUtils.STORM_HOME); |
| } |
| } |
| |
| /** |
| * Get a specific Supervisor. This is intended mostly for internal testing. |
| * |
| * @param id the id of the supervisor |
| */ |
| public synchronized Supervisor getSupervisor(String id) { |
| for (Supervisor s : supervisors) { |
| if (id.equals(s.getId())) { |
| return s; |
| } |
| } |
| return null; |
| } |
| |
| /** |
| * Kill a specific supervisor. This is intended mostly for internal testing. |
| * |
| * @param id the id of the supervisor |
| */ |
| public synchronized void killSupervisor(String id) { |
| for (Iterator<Supervisor> it = supervisors.iterator(); it.hasNext(); ) { |
| Supervisor s = it.next(); |
| if (id.equals(s.getId())) { |
| it.remove(); |
| s.close(); |
| //tmpDir will be handled separately |
| return; |
| } |
| } |
| } |
| |
| /** |
| * Add another supervisor to the topology. This is intended mostly for internal testing. |
| */ |
| public Supervisor addSupervisor() throws Exception { |
| return addSupervisor(null, null, null); |
| } |
| |
| /** |
| * Add another supervisor to the topology. This is intended mostly for internal testing. |
| * |
| * @param ports the number of ports/slots the supervisor should have |
| */ |
| public Supervisor addSupervisor(Number ports) throws Exception { |
| return addSupervisor(ports, null, null); |
| } |
| |
| /** |
| * Add another supervisor to the topology. This is intended mostly for internal testing. |
| * |
| * @param ports the number of ports/slots the supervisor should have |
| * @param id the id of the new supervisor, so you can find it later. |
| */ |
| public Supervisor addSupervisor(Number ports, String id) throws Exception { |
| return addSupervisor(ports, null, id); |
| } |
| |
| /** |
| * Add another supervisor to the topology. This is intended mostly for internal testing. |
| * |
| * @param ports the number of ports/slots the supervisor should have |
| * @param conf any config values that should be added/over written in the daemon conf of the cluster. |
| * @param id the id of the new supervisor, so you can find it later. |
| */ |
| public synchronized Supervisor addSupervisor(Number ports, Map<String, Object> conf, String id) throws Exception { |
| if (ports == null) { |
| ports = 2; |
| } |
| TmpPath tmpDir = new TmpPath(); |
| tmpDirs.add(tmpDir); |
| |
| List<Integer> portNumbers = new ArrayList<>(ports.intValue()); |
| for (int i = 0; i < ports.intValue(); i++) { |
| portNumbers.add(portCounter++); |
| } |
| |
| Map<String, Object> superConf = new HashMap<>(daemonConf); |
| if (conf != null) { |
| superConf.putAll(conf); |
| } |
| superConf.put(Config.STORM_LOCAL_DIR, tmpDir.getPath()); |
| superConf.put(DaemonConfig.SUPERVISOR_SLOTS_PORTS, portNumbers); |
| if (!Time.isSimulating()) { |
| //Monitor for assignment changes as often as possible, so e.g. shutdown happens as fast as possible. |
| superConf.put(DaemonConfig.SUPERVISOR_MONITOR_FREQUENCY_SECS, 1); |
| } |
| |
| final String superId = id == null ? Utils.uuid() : id; |
| ISupervisor isuper = new StandaloneSupervisor() { |
| @Override |
| public String generateSupervisorId() { |
| return superId; |
| } |
| }; |
| if (!ConfigUtils.isLocalMode(superConf)) { |
| throw new IllegalArgumentException("Cannot start server in distrubuted mode!"); |
| } |
| |
| Supervisor s = new Supervisor(superConf, sharedContext, isuper, metricRegistry); |
| s.launch(); |
| s.setLocalNimbus(this.nimbus); |
| this.nimbus.addSupervisor(s); |
| supervisors.add(s); |
| return s; |
| } |
| |
| private boolean areAllSupervisorsWaiting() { |
| boolean ret = true; |
| for (Supervisor s : supervisors) { |
| ret = ret && s.isWaiting(); |
| } |
| return ret; |
| } |
| |
| /** |
| * Wait for the cluster to be idle. This is intended to be used with Simulated time and is for internal testing. |
| * Note that this does not wait for spout or bolt executors to be idle. |
| * |
| * @throws InterruptedException if interrupted while waiting. |
| * @throws AssertionError if the cluster did not come to an idle point with a timeout. |
| */ |
| public void waitForIdle() throws InterruptedException { |
| waitForIdle(Testing.TEST_TIMEOUT_MS); |
| } |
| |
| /** |
| * Wait for the cluster to be idle. This is intended to be used with Simulated time and is for internal testing. |
| * Note that this does not wait for spout or bolt executors to be idle. |
| * |
| * @param timeoutMs the number of ms to wait before throwing an error. |
| * @throws InterruptedException if interrupted while waiting. |
| * @throws AssertionError if the cluster did not come to an idle point with a timeout. |
| */ |
| public void waitForIdle(long timeoutMs) throws InterruptedException { |
| Random rand = ThreadLocalRandom.current(); |
| //wait until all workers, supervisors, and nimbus is waiting |
| final long endTime = System.currentTimeMillis() + timeoutMs; |
| while (!(nimbus.isWaiting() |
| && areAllSupervisorsWaiting() |
| && areAllWorkersWaiting())) { |
| if (System.currentTimeMillis() >= endTime) { |
| LOG.info("Cluster was not idle in {} ms", timeoutMs); |
| LOG.info(Utils.threadDump()); |
| throw new AssertionError("Test timed out (" + timeoutMs + "ms) cluster not idle"); |
| } |
| Thread.sleep(rand.nextInt(20)); |
| } |
| } |
| |
| @Override |
| public void advanceClusterTime(int secs) throws InterruptedException { |
| advanceClusterTime(secs, 1); |
| } |
| |
| @Override |
| public void advanceClusterTime(int secs, int incSecs) throws InterruptedException { |
| waitForIdle(); |
| for (int amountLeft = secs; amountLeft > 0; amountLeft -= incSecs) { |
| int diff = Math.min(incSecs, amountLeft); |
| Time.advanceTimeSecs(diff); |
| waitForIdle(); |
| } |
| } |
| |
| @Override |
| public IStormClusterState getClusterState() { |
| return clusterState; |
| } |
| |
| @Override |
| public String getTrackedId() { |
| return trackId; |
| } |
| |
| //Nimbus Compatibility |
| |
| @Override |
| public void setLogConfig(String name, LogConfig config) throws TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public LogConfig getLogConfig(String name) throws TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public void debug(String name, String component, boolean enable, double samplingPercentage) |
| throws NotAliveException, AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public void setWorkerProfiler(String id, ProfileRequest profileRequest) throws TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public List<ProfileRequest> getComponentPendingProfileActions(String id, String componentId, ProfileAction action) |
| throws TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public String beginCreateBlob(String key, SettableBlobMeta meta) |
| throws AuthorizationException, KeyAlreadyExistsException, TException { |
| throw new RuntimeException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public String beginUpdateBlob(String key) throws AuthorizationException, KeyNotFoundException, TException { |
| throw new WrappedKeyNotFoundException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public void uploadBlobChunk(String session, ByteBuffer chunk) throws AuthorizationException, TException { |
| throw new RuntimeException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public void finishBlobUpload(String session) throws AuthorizationException, TException { |
| throw new RuntimeException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public void cancelBlobUpload(String session) throws AuthorizationException, TException { |
| throw new RuntimeException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public ReadableBlobMeta getBlobMeta(String key) throws AuthorizationException, KeyNotFoundException, TException { |
| throw new WrappedKeyNotFoundException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public void setBlobMeta(String key, SettableBlobMeta meta) |
| throws AuthorizationException, KeyNotFoundException, TException { |
| throw new WrappedKeyNotFoundException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public BeginDownloadResult beginBlobDownload(String key) |
| throws AuthorizationException, KeyNotFoundException, TException { |
| throw new WrappedKeyNotFoundException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public ByteBuffer downloadBlobChunk(String session) throws AuthorizationException, TException { |
| throw new RuntimeException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public void deleteBlob(String key) throws AuthorizationException, KeyNotFoundException, TException { |
| throw new WrappedKeyNotFoundException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public ListBlobsResult listBlobs(String session) throws TException { |
| //Blobs are not supported in local mode. Return nothing |
| ListBlobsResult ret = new ListBlobsResult(); |
| ret.set_keys(new ArrayList<>()); |
| return ret; |
| } |
| |
| @Override |
| public int getBlobReplication(String key) throws AuthorizationException, KeyNotFoundException, TException { |
| throw new WrappedKeyNotFoundException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public int updateBlobReplication(String key, int replication) |
| throws AuthorizationException, KeyNotFoundException, TException { |
| throw new WrappedKeyNotFoundException("BLOBS NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public void createStateInZookeeper(String key) throws TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public String beginFileUpload() throws AuthorizationException, TException { |
| //Just ignore these for now. We are going to throw it away anyways |
| return Utils.uuid(); |
| } |
| |
| @Override |
| public void uploadChunk(String location, ByteBuffer chunk) throws AuthorizationException, TException { |
| //Just throw it away in local mode |
| } |
| |
| @Override |
| public void finishFileUpload(String location) throws AuthorizationException, TException { |
| //Just throw it away in local mode |
| } |
| |
| @Override |
| public ByteBuffer downloadChunk(String id) throws AuthorizationException, TException { |
| throw new WrappedAuthorizationException("FILE DOWNLOAD NOT SUPPORTED IN LOCAL MODE"); |
| } |
| |
| @Override |
| public String getNimbusConf() throws AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public NimbusSummary getLeader() throws AuthorizationException, TException { |
| return nimbus.getLeader(); |
| } |
| |
| @Override |
| public boolean isTopologyNameAllowed(String name) throws AuthorizationException, TException { |
| return nimbus.isTopologyNameAllowed(name); |
| } |
| |
| @Override |
| public TopologyInfo getTopologyInfoWithOpts(String id, GetInfoOptions options) |
| throws NotAliveException, AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public TopologyPageInfo getTopologyPageInfo(String id, String window, boolean isIncludeSys) |
| throws NotAliveException, AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public SupervisorPageInfo getSupervisorPageInfo(String id, String host, boolean isIncludeSys) |
| throws NotAliveException, AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public ComponentPageInfo getComponentPageInfo(String topologyId, String componentId, String window, |
| boolean isIncludeSys) throws NotAliveException, AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public StormTopology getUserTopology(String id) throws NotAliveException, AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public TopologyHistoryInfo getTopologyHistory(String user) throws AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public List<OwnerResourceSummary> getOwnerResourceSummaries(String owner) throws AuthorizationException, TException { |
| // TODO Auto-generated method stub |
| throw new RuntimeException("NOT IMPLEMENTED YET"); |
| } |
| |
| @Override |
| public SupervisorAssignments getSupervisorAssignments(String node) throws AuthorizationException, TException { |
| return null; |
| } |
| |
| @Override |
| public void sendSupervisorWorkerHeartbeats(SupervisorWorkerHeartbeats heartbeats) throws AuthorizationException, TException { |
| |
| } |
| |
| @Override |
| public void sendSupervisorWorkerHeartbeat(SupervisorWorkerHeartbeat heatbeat) throws AuthorizationException, TException { |
| |
| } |
| |
| public void processWorkerMetrics(WorkerMetrics metrics) throws TException { |
| getNimbus().processWorkerMetrics(metrics); |
| } |
| |
| /** |
| * Simple way to configure a LocalCluster to meet your needs. |
| */ |
| public static class Builder { |
| private int supervisors = 2; |
| private int portsPerSupervisor = 3; |
| private Map<String, Object> daemonConf = new HashMap<>(); |
| private INimbus inimbus = null; |
| private IGroupMappingServiceProvider groupMapper = null; |
| private int supervisorSlotPortMin = 1024; |
| private boolean nimbusDaemon = false; |
| private UnaryOperator<Nimbus> nimbusWrapper = null; |
| private BlobStore store = null; |
| private TopoCache topoCache = null; |
| private IStormClusterState clusterState = null; |
| private ILeaderElector leaderElector = null; |
| private String trackId = null; |
| private boolean simulateTime = false; |
| |
| /** |
| * Set the number of supervisors the cluster should have. |
| */ |
| public Builder withSupervisors(int supervisors) { |
| if (supervisors < 0) { |
| throw new IllegalArgumentException("supervisors cannot be negative"); |
| } |
| this.supervisors = supervisors; |
| return this; |
| } |
| |
| /** |
| * Set the number of slots/ports each supervisor should have. |
| */ |
| public Builder withPortsPerSupervisor(int portsPerSupervisor) { |
| if (portsPerSupervisor < 0) { |
| throw new IllegalArgumentException("supervisor ports cannot be negative"); |
| } |
| this.portsPerSupervisor = portsPerSupervisor; |
| return this; |
| } |
| |
| /** |
| * Set the base config that the daemons should use. |
| */ |
| public Builder withDaemonConf(Map<String, Object> conf) { |
| if (conf != null) { |
| this.daemonConf = new HashMap<>(conf); |
| } |
| return this; |
| } |
| |
| /** |
| * Add an single key/value config to the daemon conf. |
| */ |
| public Builder withDaemonConf(String key, Object value) { |
| this.daemonConf.put(key, value); |
| return this; |
| } |
| |
| /** |
| * Override the INimbus instance that nimbus will use. |
| */ |
| public Builder withINimbus(INimbus inimbus) { |
| this.inimbus = inimbus; |
| return this; |
| } |
| |
| /** |
| * Override the code that maps users to groups for authorization. |
| */ |
| public Builder withGroupMapper(IGroupMappingServiceProvider groupMapper) { |
| this.groupMapper = groupMapper; |
| return this; |
| } |
| |
| /** |
| * When assigning ports to worker slots start at minPort. |
| */ |
| public Builder withSupervisorSlotPortMin(Number minPort) { |
| int port = 1024; |
| if (minPort == null) { |
| LOG.warn("Number is null... {}", minPort); |
| } else { |
| port = minPort.intValue(); |
| } |
| if (port <= 0) { |
| throw new IllegalArgumentException("port must be positive"); |
| } |
| this.supervisorSlotPortMin = port; |
| return this; |
| } |
| |
| /** |
| * Have the local nimbus actually launch a thrift server. This is intended to be used mostly for internal storm testing. |
| */ |
| public Builder withNimbusDaemon() { |
| return withNimbusDaemon(true); |
| } |
| |
| /** |
| * If nimbusDaemon is true the local nimbus will launch a thrift server. This is intended to be used mostly for internal storm |
| * testing. |
| */ |
| public Builder withNimbusDaemon(Boolean nimbusDaemon) { |
| if (nimbusDaemon == null) { |
| nimbusDaemon = false; |
| LOG.warn("nimbusDaemon is null"); |
| } |
| this.nimbusDaemon = nimbusDaemon; |
| return this; |
| } |
| |
| /** |
| * Turn on simulated time in the cluster. This allows someone to simulate long periods of time for timeouts etc when testing |
| * nimbus/supervisors themselves. NOTE: that this only works for code that uses the {@link org.apache.storm.utils.Time} class for |
| * time management so it will not work in all cases. |
| */ |
| public Builder withSimulatedTime() { |
| return withSimulatedTime(true); |
| } |
| |
| /** |
| * Turn on simulated time in the cluster. This allows someone to simulate long periods of time for timeouts etc when testing |
| * nimbus/supervisors themselves. NOTE: that this only works for code that uses the {@link org.apache.storm.utils.Time} class for |
| * time management so it will not work in all cases. |
| */ |
| public Builder withSimulatedTime(boolean simulateTime) { |
| this.simulateTime = simulateTime; |
| return this; |
| } |
| |
| /** |
| * Before nimbus is created/used call nimbusWrapper on it first and use the result instead. This is intended for internal testing |
| * only, and it here to allow a mocking framework to spy on the nimbus class. |
| */ |
| public Builder withNimbusWrapper(UnaryOperator<Nimbus> nimbusWrapper) { |
| this.nimbusWrapper = nimbusWrapper; |
| return this; |
| } |
| |
| /** |
| * Use the following blobstore instead of the one in the config. This is intended mostly for internal testing with Mocks. |
| */ |
| public Builder withBlobStore(BlobStore store) { |
| this.store = store; |
| return this; |
| } |
| |
| /** |
| * Use the following topo cache instead of creating out own. This is intended mostly for internal testing with Mocks. |
| */ |
| public Builder withTopoCache(TopoCache topoCache) { |
| this.topoCache = topoCache; |
| return this; |
| } |
| |
| /** |
| * Use the following clusterState instead of the one in the config. This is intended mostly for internal testing with Mocks. |
| */ |
| public Builder withClusterState(IStormClusterState clusterState) { |
| this.clusterState = clusterState; |
| return this; |
| } |
| |
| /** |
| * Use the following leaderElector instead of the one in the config. This is intended mostly for internal testing with Mocks. |
| */ |
| public Builder withLeaderElector(ILeaderElector leaderElector) { |
| this.leaderElector = leaderElector; |
| return this; |
| } |
| |
| /** |
| * A tracked cluster can run tracked topologies. See {@link org.apache.storm.testing.TrackedTopology} for more information on |
| * tracked topologies. |
| * |
| * @param trackId an arbitrary unique id that is used to keep track of tracked topologies |
| */ |
| public Builder withTracked(String trackId) { |
| this.trackId = trackId; |
| return this; |
| } |
| |
| /** |
| * A tracked cluster can run tracked topologies. See {@link org.apache.storm.testing.TrackedTopology} for more information on |
| * tracked topologies. |
| */ |
| public Builder withTracked() { |
| this.trackId = Utils.uuid(); |
| return this; |
| } |
| |
| /** |
| * Builds a new LocalCluster. |
| * |
| * @return the LocalCluster |
| * |
| * @throws Exception on any one of many different errors. This is intended for testing so yes it is ugly and throws Exception... |
| */ |
| public LocalCluster build() throws Exception { |
| return new LocalCluster(this); |
| } |
| } |
| |
| private static class TrackedStormCommon extends StormCommon { |
| |
| private final String id; |
| |
| TrackedStormCommon(String id) { |
| this.id = id; |
| } |
| |
| @Override |
| public IBolt makeAckerBoltImpl() { |
| return new NonRichBoltTracker(new Acker(), id); |
| } |
| } |
| |
| /** |
| * When running a topology locally, for tests etc. It is helpful to be sure that the topology is dead before the test exits. This is |
| * an AutoCloseable topology that not only gives you access to the compiled StormTopology but also will kill the topology when it |
| * closes. |
| * |
| * <code> |
| * try (LocalTopology testTopo = cluster.submitTopology("testing", ...)) { |
| * // Run Some test |
| * } // The topology has been killed |
| * </code> |
| */ |
| public class LocalTopology extends StormTopology implements ILocalTopology { |
| private static final long serialVersionUID = 6145919776650637748L; |
| private final String topoName; |
| |
| public LocalTopology(String topoName, StormTopology topo) { |
| super(topo); |
| this.topoName = topoName; |
| } |
| |
| @Override |
| public void close() throws TException { |
| killTopologyWithOpts(topoName, KILL_NOW); |
| } |
| } |
| } |
