create and start a supervisor

@param conf          : configuration (default.yaml & storm.yaml)
@param sharedContext : null (right now)
@return SupervisorManger: which is used to shutdown all workers and supervisor
507740
@SuppressWarnings("rawtypes")
    public SupervisorManger mkSupervisor(Map conf, IContext sharedContext) throws Exception {

        LOG.info("Starting Supervisor with conf " + conf);

        /**
         * Step 1: cleanup all files in /storm-local-dir/supervisor/tmp
         */
        String path = StormConfig.supervisorTmpDir(conf);
        FileUtils.cleanDirectory(new File(path));

        /**
         * Step 2: create ZK operation instance StormClusterState
         */

        StormClusterState stormClusterState = Cluster.mk_storm_cluster_state(conf);

        String hostName = JStormServerUtils.getHostName(conf);
        WorkerReportError workerReportError = new WorkerReportError(stormClusterState, hostName);

        /**
         * Step 3, create LocalStat (a simple KV store)
         * 3.1 create LocalState instance;
         * 3.2 get supervisorId, if there's no supervisorId, create one
         */

        LocalState localState = StormConfig.supervisorState(conf);

        String supervisorId = (String) localState.get(Common.LS_ID);
        if (supervisorId == null) {
            supervisorId = UUID.randomUUID().toString();
            localState.put(Common.LS_ID, supervisorId);
        }
        //clean LocalStat's zk-assignment & versions
        localState.remove(Common.LS_LOCAl_ZK_ASSIGNMENTS);
        localState.remove(Common.LS_LOCAL_ZK_ASSIGNMENT_VERSION);

        Vector<AsyncLoopThread> threads = new Vector<>();

        // Step 4 create HeartBeat
        // every supervisor.heartbeat.frequency.secs, write SupervisorInfo to ZK
        // sync heartbeat to nimbus
        Heartbeat hb = new Heartbeat(conf, stormClusterState, supervisorId, localState);
        hb.update();

        AsyncLoopThread heartbeat = new AsyncLoopThread(hb, false, null, Thread.MIN_PRIORITY, true);
        threads.add(heartbeat);

        // Sync heartbeat to Apsara Container
        AsyncLoopThread syncContainerHbThread = SyncContainerHb.mkSupervisorInstance(conf);
        if (syncContainerHbThread != null) {
            threads.add(syncContainerHbThread);
        }

        // Step 5 create and start sync Supervisor thread
        // every supervisor.monitor.frequency.secs second run SyncSupervisor
        ConcurrentHashMap<String, String> workerThreadPids = new ConcurrentHashMap<>();
        SyncProcessEvent syncProcessEvent = new SyncProcessEvent(
                supervisorId, conf, localState, workerThreadPids, sharedContext, workerReportError, stormClusterState);

        EventManagerImp syncSupEventManager = new EventManagerImp();
        AsyncLoopThread syncSupEventThread = new AsyncLoopThread(syncSupEventManager);
        threads.add(syncSupEventThread);

        SyncSupervisorEvent syncSupervisorEvent = new SyncSupervisorEvent(
                supervisorId, conf, syncSupEventManager, stormClusterState, localState, syncProcessEvent, hb);

        int syncFrequency = JStormUtils.parseInt(conf.get(Config.SUPERVISOR_MONITOR_FREQUENCY_SECS));
        EventManagerPusher syncSupervisorPusher = new EventManagerPusher(
                syncSupEventManager, syncSupervisorEvent, syncFrequency);
        AsyncLoopThread syncSupervisorThread = new AsyncLoopThread(syncSupervisorPusher);
        threads.add(syncSupervisorThread);

        // Step 6 start httpserver
        Httpserver httpserver = null;
        if (!StormConfig.local_mode(conf)) {
            int port = ConfigExtension.getSupervisorDeamonHttpserverPort(conf);
            httpserver = new Httpserver(port, conf);
            httpserver.start();
        }

        //Step 7 check supervisor
        if (!StormConfig.local_mode(conf)) {
            if (ConfigExtension.isEnableCheckSupervisor(conf)) {
                SupervisorHealth supervisorHealth = new SupervisorHealth(conf, hb, supervisorId);
                AsyncLoopThread healthThread = new AsyncLoopThread(supervisorHealth, false, null, Thread.MIN_PRIORITY, true);
                threads.add(healthThread);
            }

            // init refresh config thread
            AsyncLoopThread refreshConfigThread = new AsyncLoopThread(new SupervisorRefreshConfig(conf));
            threads.add(refreshConfigThread);
        }

        // create SupervisorManger which can shutdown all supervisor and workers
        return new SupervisorManger(
                conf, supervisorId, threads, syncSupEventManager, httpserver, stormClusterState, workerThreadPids);
    }
213176
def start(self):
        '''
        
        '''
        super(Minion, self).start()
        while True:
            try:
                self._real_start()
            except SaltClientError as exc:
                # Restart for multi_master failover when daemonized
                if self.options.daemon:
                    continue
            break
1265795
protected ProducerCloseable<String, byte[]> createProducer(ProducerConfig config) {
    return this.closer.register(new ProducerCloseable<String, byte[]>(config));
  }
1299858
def create_process_worker(self, cmd_list, environ=None):
        """"""
        worker = ProcessWorker(cmd_list, environ=environ)
        self._create_worker(worker)
        return worker
64138
function createWorkerExitHandler(port) {
        return function (code, signal) {
            var workerIndex = port - startPort;

            workers[workerIndex] = null;

            if (code !== 0 || code === null) {
                console.log('Worker exited with code: ' + code);

                if (!shuttingDown) {
                    // Start worker again after 1 sec
                    setTimeout(function () {
                        if (!workers[workerIndex]) {
                            workers[workerIndex] = startWorker(port);
                        }
                    }, 1000);
                }
            }
        };
    }
744249
public CuratorFramework newManagedCurator(LifecycleEnvironment env) {
        CuratorFramework curator = newCurator();
        env.manage(new ManagedCuratorFramework(curator));
        return curator;
    }
1236374
def _init_supervisor_rpc(self, rpc_or_port):
        '''

        '''
        if isinstance(rpc_or_port, int):
            if self.username:
                leader = 'http://{self.username}:{self.password}@'
            else:
                leader = 'http://'
            tmpl = leader + '{self.name}:{port}'
            url = tmpl.format(self=self, port=rpc_or_port)
            self.rpc = xmlrpc_client.ServerProxy(
                url, transport=TimeoutTransport())
        else:
            self.rpc = rpc_or_port
        self.supervisor = self.rpc.supervisor
315908
public static void main(String[] args) {
    if (args.length != 0) {
      LOG.info("java -cp {} {}", RuntimeConstants.ALLUXIO_JAR,
          AlluxioWorker.class.getCanonicalName());
      System.exit(-1);
    }

    if (!ConfigurationUtils.masterHostConfigured(ServerConfiguration.global())) {
      ProcessUtils.fatalError(LOG,
          ConfigurationUtils.getMasterHostNotConfiguredMessage("Alluxio worker"));
    }

    CommonUtils.PROCESS_TYPE.set(CommonUtils.ProcessType.WORKER);
    MasterInquireClient masterInquireClient =
        MasterInquireClient.Factory.create(ServerConfiguration.global());
    try {
      RetryUtils.retry("load cluster default configuration with master", () -> {
        InetSocketAddress masterAddress = masterInquireClient.getPrimaryRpcAddress();
        ServerConfiguration.loadClusterDefaults(masterAddress);
      }, RetryUtils.defaultWorkerMasterClientRetry(
          ServerConfiguration.getDuration(PropertyKey.WORKER_MASTER_CONNECT_RETRY_TIMEOUT)));
    } catch (IOException e) {
      ProcessUtils.fatalError(LOG,
          "Failed to load cluster default configuration for worker: %s", e.getMessage());
    }
    WorkerProcess process = WorkerProcess.Factory.create();
    ProcessUtils.run(process);
  }
1065660
def __create_supervisor_session(self, supervisor):
        
        session_params = {
            'forceLogoutSession': self.force_logout_session,
            'rollingPeriod': self.rolling_period,
            'statisticsRange': self.statistics_range,
            'shiftStart': self.__to_milliseconds(
                self.shift_start_hour,
            ),
            'timeZone': self.__to_milliseconds(
                self.time_zone_offset,
            ),
        }
        supervisor.setSessionParameters(session_params)
        return session_params
1077625
public function runAndReload()
    {

        // start supervisor and reload configuration
        $commands = [];
        $commands[] = sprintf(' && %s', $this->createSupervisorControlCommand('reread'));
        $commands[] = sprintf(' && %s', $this->createSupervisorControlCommand('update'));
        $this->run(implode('', $commands));
    }
248610
public CuratorFramework mkClient(Map conf, List<String> servers, Object port,
                                     String root, final WatcherCallBack watcher) {

        CuratorFramework fk = Utils.newCurator(conf, servers, port, root);

        fk.getCuratorListenable().addListener(new CuratorListener() {
            @Override
            public void eventReceived(CuratorFramework _fk, CuratorEvent e) throws Exception {
                if (e.getType().equals(CuratorEventType.WATCHED)) {
                    WatchedEvent event = e.getWatchedEvent();

                    watcher.execute(event.getState(), event.getType(), event.getPath());
                }

            }
        });

        fk.getUnhandledErrorListenable().addListener(new UnhandledErrorListener() {
            @Override
            public void unhandledError(String msg, Throwable error) {
                String errmsg = "Unrecoverable zookeeper error, halting process: " + msg;
                LOG.error(errmsg, error);
                JStormUtils.halt_process(1, "Unrecoverable zookeeper error");

            }
        });
        fk.start();
        return fk;
    }
245059
def upload_supervisor_app_conf(app_name, template_name=None, context=None):
    """"""

    default = {'app_name': app_name}
    context = context or {}
    default.update(context)
    template_name = template_name or [u'supervisor/%s.conf' % app_name, u'supervisor/base.conf']
    destination = u'/etc/supervisor/conf.d/%s.conf' % app_name
    upload_template(template_name, destination, context=default, use_sudo=True)
    supervisor_command(u'update')
1092900
private Supervisor createSupervisor(final Job job, final Map<String, Integer> portAllocation) {
    log.debug("creating job supervisor: {}", job);
    final TaskStatus taskStatus = model.getTaskStatus(job.getId());
    final String containerId = (taskStatus == null) ? null : taskStatus.getContainerId();
    final Supervisor supervisor = supervisorFactory.create(job, containerId, portAllocation,
        supervisorListener);
    supervisors.put(job.getId(), supervisor);
    return supervisor;
  }
836653
public CompletableFuture<Acknowledge> start(final JobMasterId newJobMasterId) throws Exception {
		// make sure we receive RPC and async calls
		start();

		return callAsyncWithoutFencing(() -> startJobExecution(newJobMasterId), RpcUtils.INF_TIMEOUT);
	}
1201764
def create_manager(self, yosai, settings, session_attributes):
        
        mgr_settings = SecurityManagerSettings(settings)
        attributes = mgr_settings.attributes

        realms = self._init_realms(settings, attributes['realms'])

        session_attributes = self._init_session_attributes(session_attributes, attributes)

        serialization_manager =\
            SerializationManager(session_attributes,
                                 serializer_scheme=attributes['serializer'])

        # the cache_handler doesn't initialize a cache_realm until it gets
        # a serialization manager, which is assigned within the SecurityManager
        cache_handler = self._init_cache_handler(settings,
                                                 attributes['cache_handler'],
                                                 serialization_manager)

        manager = mgr_settings.security_manager(yosai,
                                                settings,
                                                realms=realms,
                                                cache_handler=cache_handler,
                                                serialization_manager=serialization_manager)

        return manager
1322268
public JobStatusListener createActivatorDeactivator() {
		synchronized (lock) {
			if (shutdown) {
				throw new IllegalArgumentException("Checkpoint coordinator is shut down");
			}

			if (jobStatusListener == null) {
				jobStatusListener = new CheckpointCoordinatorDeActivator(this);
			}

			return jobStatusListener;
		}
	}
468898
private void setup(final Handler<AsyncResult<Void>> doneHandler) {
    // Retrieve the component context from the coordinator (the current cluster).
    // If the context has changed due to a network configuration change, the
    // internal context and input/output connections will be automatically updated.
    log.debug(String.format("%s - Starting cluster coordination", DefaultComponent.this));
    coordinator.start(new Handler<AsyncResult<Void>>() {
      @Override
      public void handle(AsyncResult<Void> result) {
        if (result.failed()) {
          new DefaultFutureResult<Void>(result.cause()).setHandler(doneHandler);
        } else {
          // We have to make sure the input and output collectors are started
          // simultaneously in order to support circular connections. If both
          // input and output aren't started at the same time then circular
          // connections will never open.
          final CountingCompletionHandler<Void> ioHandler = new CountingCompletionHandler<Void>(2);
          ioHandler.setHandler(new Handler<AsyncResult<Void>>() {
            @Override
            public void handle(AsyncResult<Void> result) {
              if (result.failed()) {
                new DefaultFutureResult<Void>(result.cause()).setHandler(doneHandler);
              } else {
                // Tell the coordinator we're ready for the network to start.
                coordinator.resume();
              }
            }
          });

          output.open(new Handler<AsyncResult<Void>>() {
            @Override
            public void handle(AsyncResult<Void> result) {
              if (result.failed()) {
                log.error(String.format("%s - Failed to open component outputs", DefaultComponent.this), result.cause());
                ioHandler.fail(result.cause());
              } else {
                ioHandler.succeed();
              }
            }
          });

          input.open(new Handler<AsyncResult<Void>>() {
            @Override
            public void handle(AsyncResult<Void> result) {
              if (result.failed()) {
                log.error(String.format("%s - Failed to open component inputs", DefaultComponent.this), result.cause());
                ioHandler.fail(result.cause());
              } else {
                ioHandler.succeed();
              }
            }
          });
        }
      }
    });

    // The resume handler will be called by the coordinator once the
    // network's manager has indicated that all the components in the
    // network have finished setting up their connections.
    coordinator.resumeHandler(new Handler<Void>() {
      @Override
      @SuppressWarnings("unchecked")
      public void handle(Void _) {
        if (!started) {
          started = true;
          log.debug(String.format("%s - Started", DefaultComponent.this, context.component().name(), context.number()));
          List<ComponentHook> hooks = context.component().hooks();
          for (ComponentHook hook : hooks) {
            hook.handleStart(DefaultComponent.this);
          }
          new DefaultFutureResult<Void>((Void) null).setHandler(doneHandler);
        }
      }
    });
  }