code: /*
   * Copyright © 2014-2017 EntIT Software LLC, a Micro Focus company (L.P.)
   *
   * Licensed 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 io.cloudslang.worker.management.services;
  
  import io.cloudslang.engine.node.entities.WorkerKeepAliveInfo;
  import io.cloudslang.engine.node.services.WorkerNodeService;
  import io.cloudslang.orchestrator.services.EngineVersionService;
  import io.cloudslang.worker.management.WorkerConfigurationService;
  import io.cloudslang.worker.management.monitor.WorkerStateUpdateService;
  import io.cloudslang.worker.management.queue.WorkerQueueDetailsContainer;
  import io.cloudslang.worker.management.queue.WorkerQueueDetailsHolder;
  import org.apache.commons.lang.ArrayUtils;
  import org.apache.logging.log4j.LogManager;
  import org.apache.logging.log4j.Logger;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.beans.factory.annotation.Qualifier;
  import org.springframework.context.ApplicationEvent;
  import org.springframework.context.ApplicationListener;
  import org.springframework.context.event.ContextClosedEvent;
  import org.springframework.context.event.ContextRefreshedEvent;
  
  import javax.annotation.PostConstruct;
  import javax.annotation.Resource;
  import java.io.File;
  import java.io.FileFilter;
  import java.util.Collection;
  import java.util.LinkedList;
  import java.util.Queue;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Future;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  
  import static ch.lambdaj.Lambda.max;
  import static ch.lambdaj.Lambda.on;
  import static java.lang.Boolean.FALSE;
  import static java.lang.Boolean.parseBoolean;
  import static java.lang.String.valueOf;
  import static java.lang.System.getProperty;
  
  
  public class WorkerManager implements ApplicationListener, EndExecutionCallback, WorkerRecoveryListener {
  
      private static final Logger logger = LogManager.getLogger(WorkerManager.class);
      private static final int KEEP_ALIVE_FAIL_LIMIT = 5;
      private static final String DOTNET_PATH = System.getenv("WINDIR") + "/Microsoft.NET/Framework";
  
      @Resource
      private String workerUuid;
  
      @Autowired
      protected WorkerNodeService workerNodeService;
  
      @Autowired
      private EngineVersionService engineVersionService;
  
      @Autowired
      protected WorkerConfigurationService workerConfigurationService;
  
      @Autowired
      protected WorkerRecoveryManager recoveryManager;
  
      @Autowired
      protected WorkerVersionService workerVersionService;
  
      private BlockingQueue<Runnable> inBuffer;
  
      @Autowired
      @Qualifier("numberOfExecutionThreads")
      private Integer numberOfThreads;
  
      @Autowired(required = false)
      @Qualifier("initStartUpSleep")
      private Long initStartUpSleep = 15 * 1000L; // by default 15 seconds
  
      @Autowired(required = false)
      @Qualifier("maxStartUpSleep")
      private Long maxStartUpSleep = 10 * 60 * 1000L; // by default 10 minutes
  
      @Autowired
      private WorkerConfigurationUtils workerConfigurationUtils;
  
      @Autowired
      @Qualifier("inBufferCapacity")
      private Integer capacity;
  
      @Autowired
      private WorkerStateUpdateService workerStateUpdateService;
  
      @Autowired
      private WorkerQueueDetailsContainer workerQueueDetailsContainer;
  
      private int keepAliveFailCount = 0;
  
      private ExecutorService executorService;
  
      private ConcurrentMap<Long, Queue<Future>> mapOfRunningTasks;
  
      private volatile boolean endOfInit = false;
  
      private volatile boolean initStarted = false;
  
      private boolean up = false;
  
      private boolean versionMismatch = false;
  
      private int threadPoolVersion = 0;
  
      private boolean newCancelBehaviour;
  
      @PostConstruct
      private void init() {
          logger.info("Initialize worker with UUID: " + workerUuid);
          System.setProperty("worker.uuid", workerUuid); //do not remove!!!
  
          inBuffer = workerConfigurationUtils.getBlockingQueue(numberOfThreads, capacity);
  
          executorService = new ThreadPoolExecutor(numberOfThreads,
                  numberOfThreads,
                  Long.MAX_VALUE, TimeUnit.NANOSECONDS,
                  inBuffer,
                  new WorkerThreadFactory(valueOf(incrementAndGetTreadPoolVersion()) + "_WorkerExecutionThread"));
  
          mapOfRunningTasks = new ConcurrentHashMap<>(numberOfThreads);
          newCancelBehaviour = parseBoolean(getProperty("enable.new.cancel.execution", FALSE.toString()));
      }
  
      public void addExecution(long executionId, Runnable runnable) {
          // Since we can offer to thread pool queue from SimpleExecutionRunnable run method
          // it is possible we will have step x + 1 of an execution plan that is in the map of running tasks,
          // but step x did not yet clean itself from the map
          Future future = executorService.submit(runnable);
          mapOfRunningTasks.merge(executionId, newQueue(future), this::addLists);
      }
  
      private Queue<Future> newQueue(Future future) {
          Queue<Future> queue = new LinkedList<>();
          queue.offer(future);
          return queue;
      }
  
      private Queue<Future> addLists(Queue<Future> oldValue, Queue<Future> newValue) {
          oldValue.offer(newValue.poll()); // There is only one value in newValue
          return oldValue;
      }
  
      @Override
      public void endExecution(long executionId) {
          mapOfRunningTasks.merge(executionId, new LinkedList<>(),
                  (queue, newValue) -> {
                      queue.poll();
                      return !queue.isEmpty() ? queue : null;
                  }
          );
      }
  
      public int getInBufferSize() {
          return inBuffer.size();
      }
  
      @SuppressWarnings("unused")
      // Scheduled in scoreWorkerSchedulerContext.xml
      public void interruptCanceledExecutions() {
          try {
              if (!newCancelBehaviour) { // For backward compatibility
                  for (Long executionId : mapOfRunningTasks.keySet()) {
                      if (workerConfigurationService.isExecutionCancelled(executionId)) {
                          Collection<Future> futures = mapOfRunningTasks.get(executionId);
                          for (Future future : futures) {
                              future.cancel(true);
                          }
                      }
                  }
              }
          } catch (Exception exc) {
              logger.error("Could not stop cancelled executions: ", exc);
          }
      }
  
      @SuppressWarnings("unused")
      // Scheduled in xml
      public void workerKeepAlive() {
          if (!recoveryManager.isInRecovery()) {
              if (endOfInit) {
                  WorkerQueueDetailsHolder queueDetailsHolder = null;
                  try {
                      WorkerKeepAliveInfo workerKeepAliveInfo = workerNodeService.newKeepAlive(workerUuid, versionMismatch);
                      String newWrv = workerKeepAliveInfo.getWorkerRecoveryVersion();
                      workerStateUpdateService.setEnableState(workerKeepAliveInfo.isActive());
                      workerStateUpdateService.setMonitoringState(workerKeepAliveInfo.shouldMonitor());
                      String currentWrv = recoveryManager.getWRV();
                      queueDetailsHolder = new WorkerQueueDetailsHolder(workerKeepAliveInfo.getQueueDetails());
                      //do not update it!!! if it is different than we have - restart worker (clean state)
                      if (!currentWrv.equals(newWrv)) {
                          logger.warn("Got new WRV from Orchestrator during keepAlive(). Going to reload...");
                          recoveryManager.doRecovery();
                      }
                      keepAliveFailCount = 0;
                  } catch (Exception e) {
                      keepAliveFailCount++;
                      logger.error("Could not send keep alive to Central, keepAliveFailCount = " + keepAliveFailCount, e);
                      if (keepAliveFailCount >= KEEP_ALIVE_FAIL_LIMIT) {
                          logger.error("Failed sending keepAlive for " + KEEP_ALIVE_FAIL_LIMIT
                                  + " times. Invoking worker internal recovery...");
                          recoveryManager.doRecovery();
                      }
                  } finally {
                      if (queueDetailsHolder != null) {
                          workerQueueDetailsContainer.setQueueConfiguration(queueDetailsHolder);
                      }
                  }
              }
          } else {
              if (logger.isDebugEnabled()) {
                  logger.debug("worker waits for recovery");
              }
          }
      }
  
      @SuppressWarnings("unused") // called by scheduler
      public void logStatistics() {
          if (logger.isDebugEnabled()) {
              logger.debug("InBuffer size: " + getInBufferSize());
              logger.debug("Running task size: " + mapOfRunningTasks.size());
          }
      }
  
      public String getWorkerUuid() {
          return workerUuid;
      }
  
      public int getRunningTasksCount() {
          return mapOfRunningTasks.size();
      }
  
      public int getExecutionThreadsCount() {
          return numberOfThreads;
      }
  
      @Override
      public void onApplicationEvent(final ApplicationEvent applicationEvent) {
          if (applicationEvent instanceof ContextRefreshedEvent && !initStarted) {
              doStartup();
          } else if (applicationEvent instanceof ContextClosedEvent) {
              doShutdown();
          }
      }
  
      private void doStartup() {
          new Thread(new Runnable() {
              @Override
              public void run() {
                  versionMismatch = !workerVersionService.getWorkerVersionId().equals(engineVersionService.getEngineVersionId());
                  initStarted = true;
                  long sleep = initStartUpSleep;
                  boolean shouldRetry = true;
                  while (shouldRetry) {
                      try {
                          String newWrv = workerNodeService.up(workerUuid, workerVersionService.getWorkerVersion(),
                                  workerVersionService.getWorkerVersionId(), versionMismatch);
                          recoveryManager
                                  .setWRV(newWrv); //we do set of WRV here and in doRecovery() only!!! not in keepalive!!!
                          shouldRetry = false;
                          logger.info("Worker is up");
                      } catch (Exception ex) {
                          logger.error("Worker failed on start up, will retry in a " + sleep / 1000 + " seconds", ex);
                          try {
                              Thread.sleep(sleep);
                          } catch (InterruptedException iex) {/*do nothing*/}
                          sleep = Math.min(maxStartUpSleep, sleep * 2); // double the sleep time until max 10 minute
                      }
                  }
  
                  endOfInit = true;
  
                  //Check that this Worker is in the same version as engine - if not - stay idle
                  if (!versionMismatch) {
  
                      //mark that worker is up and its recovery is ended - only now we can start asking for messages from queue
                      up = true;
                      workerConfigurationService.setEnabled(true);
                      workerNodeService.updateEnvironmentParams(workerUuid,
                              getProperty("os.name"),
                              getProperty("java.version"),
                              resolveDotNetVersion());
                  } else {
                      logger.warn(
                              "Worker's version is not equal to engine version. Won't be able to start processing flows!");
                  }
              }
          }).start();
      }
  
      private void doShutdown() {
          endOfInit = false;
          initStarted = false;
          workerConfigurationService.setEnabled(false);
          up = false;
          logger.info("The worker is down");
      }
  
      protected static String resolveDotNetVersion() {
          File dotNetHome = new File(DOTNET_PATH);
          if (dotNetHome.isDirectory()) {
              File[] versionFolders = dotNetHome.listFiles(new FileFilter() {
  
                  @Override
                  public boolean accept(File file) {
                      return file.isDirectory() && file.getName().startsWith("v");
                  }
              });
              if (!ArrayUtils.isEmpty(versionFolders)) {
                  String maxVersion = max(versionFolders, on(File.class).getName()).substring(1);
                  return maxVersion.substring(0, 1) + ".x";
              }
          }
          return "N/A";
      }
  
      public boolean isUp() {
          return up;
      }
  
      public synchronized boolean isFromCurrentThreadPool(String threadName) {
          // Since this is executed in the score threadpool, making code simpler to avoid unnecessary contention
          return threadName.startsWith(valueOf(threadPoolVersion));
      }
  
      //Must clean the buffer that holds Runnables that wait for execution and also drop all the executions that currently run
      public void doRecovery() {
  
  //        Attempts to stop all actively executing tasks, halts the
  //        processing of waiting tasks, and returns a list of the tasks
  //        that were awaiting execution.
  //
  //        This method does not wait for actively executing tasks to
  //        terminate.
  //
  //        There are no guarantees beyond best-effort attempts to stop
  //        processing actively executing tasks.  For example, typical
  //        implementations will cancel via {@link Thread#interrupt}, so any
  //        task that fails to respond to interrupts may never terminate.
          try {
              synchronized (this) {
                  executorService.shutdownNow(); //shutting down current running threads
                  threadPoolVersion++;           //updating the thread pool version to a new one - so current running threads will exit
                  logger.warn(
                          "Worker is in doRecovery(). Cleaning state and cancelling running tasks. It may take up to 30 seconds...");
              }
  
              boolean finished = executorService.awaitTermination(30, TimeUnit.SECONDS);
  
              if (finished) {
                  logger.warn("Worker succeeded to cancel running tasks during doRecovery().");
              } else {
                  logger.warn("Not all running tasks responded to cancel.");
              }
          } catch (InterruptedException ex) {/*ignore*/}
  
          mapOfRunningTasks.clear();
  
          //Make new executor
          executorService = new ThreadPoolExecutor(numberOfThreads,
                  numberOfThreads,
                  Long.MAX_VALUE, TimeUnit.NANOSECONDS,
                  inBuffer,
                  new WorkerThreadFactory(valueOf(getTreadPoolVersion()) + "_WorkerExecutionThread"));
      }
  
      private synchronized int getTreadPoolVersion() {
          return threadPoolVersion;
      }
  
      private synchronized int incrementAndGetTreadPoolVersion() {
          return (++threadPoolVersion);
      }
  
  }
  