/*
 * 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.uima.ducc.agent.event;

import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.camel.Body;
import org.apache.uima.ducc.agent.Agent;
import org.apache.uima.ducc.agent.NodeAgent;
import org.apache.uima.ducc.agent.ProcessLifecycleController;
import org.apache.uima.ducc.agent.deploy.DuccWorkHelper;
import org.apache.uima.ducc.common.head.IDuccHead.DuccHeadState;
import org.apache.uima.ducc.common.utils.DuccLogger;
import org.apache.uima.ducc.common.utils.Utils;
import org.apache.uima.ducc.common.utils.id.DuccId;
import org.apache.uima.ducc.transport.agent.ProcessStateUpdate;
import org.apache.uima.ducc.transport.cmdline.ICommandLine;
import org.apache.uima.ducc.transport.dispatcher.DuccEventDispatcher;
import org.apache.uima.ducc.transport.event.DuccJobsStateEvent;
import org.apache.uima.ducc.transport.event.ProcessPurgeDuccEvent;
import org.apache.uima.ducc.transport.event.ProcessStartDuccEvent;
import org.apache.uima.ducc.transport.event.ProcessStateUpdateDuccEvent;
import org.apache.uima.ducc.transport.event.ProcessStopDuccEvent;
import org.apache.uima.ducc.transport.event.common.DuccUserReservation;
import org.apache.uima.ducc.transport.event.common.DuccWorkPopDriver;
import org.apache.uima.ducc.transport.event.common.IDuccJobDeployment;
import org.apache.uima.ducc.transport.event.common.IDuccProcess;
import org.apache.uima.ducc.transport.event.common.IDuccProcessType.ProcessType;
import org.apache.uima.ducc.transport.event.common.IDuccTypes.DuccType;
import org.apache.uima.ducc.transport.event.common.IDuccWork;
import org.apache.uima.ducc.transport.event.common.IDuccWorkExecutable;
import org.apache.uima.ducc.transport.event.common.IDuccWorkJob;
import org.apache.uima.ducc.transport.event.common.IProcessState.ProcessState;
import org.apache.uima.ducc.transport.event.delegate.DuccEventDelegateListener;
import org.springframework.beans.factory.annotation.Qualifier;

@Qualifier(value = "ael")

public class AgentEventListener implements DuccEventDelegateListener {
  DuccLogger logger = DuccLogger.getLogger(this.getClass(), Agent.COMPONENT_NAME);

  DuccId jobid = null;

  ProcessLifecycleController lifecycleController = null;

  // On startup of the Agent we may need to do cleanup of cgroups.
  // This cleanup will happen once right after processing of the first OR publication.
  // private boolean cleanupPhase = true;
  private AtomicLong lastSequence = new AtomicLong();

  private volatile boolean forceInventoryUpdateDueToSequence = false;

  private NodeAgent agent;

  private static DuccWorkHelper dwHelper = null;

  public AgentEventListener(NodeAgent agent, ProcessLifecycleController lifecycleController) {
    this.agent = agent;
    this.lifecycleController = lifecycleController;
    dwHelper = new DuccWorkHelper();
  }

  public AgentEventListener(NodeAgent agent) {
    this(agent, null);
    // this.agent = agent;
  }

  public void setDuccEventDispatcher(DuccEventDispatcher eventDispatcher) {
    // this.eventDispatcher = eventDispatcher;
  }

  private void reportIncomingStateForThisNode(DuccJobsStateEvent duccEvent) throws Exception {
    StringBuffer sb = new StringBuffer();
    for (IDuccJobDeployment jobDeployment : duccEvent.getJobList()) {
      if (isTargetNodeForProcess(jobDeployment.getJdProcess())) {
        IDuccProcess process = jobDeployment.getJdProcess();
        sb.append("\nJD--> JobId:" + jobDeployment.getJobId() + " ProcessId:" + process.getDuccId()
                + " PID:" + process.getPID() + " Status:" + process.getProcessState()
                + " Resource State:" + process.getResourceState() + " isDeallocated:"
                + process.isDeallocated());
      }
      for (IDuccProcess process : jobDeployment.getJpProcessList()) {
        if (isTargetNodeForProcess(process)) {
          sb.append("\n\tJob ID:" + jobDeployment.getJobId() + " ProcessId:" + process.getDuccId()
                  + " PID:" + process.getPID() + " Status:" + process.getProcessState()
                  + " Resource State:" + process.getResourceState() + " isDeallocated:"
                  + process.isDeallocated());
        }
      }
    }
    if (sb.length() > 0) {
      logger.info("reportIncomingStateForThisNode", null, sb.toString());
    }

  }

  public boolean forceInvotoryUpdate() {
    return forceInventoryUpdateDueToSequence;
  }

  public void resetForceInventoryUpdateFlag() {
    forceInventoryUpdateDueToSequence = false;
  }

  private void handleJobOrPOPState(IDuccJobDeployment jobDeployment, IDuccWork dw,
          IDuccProcess process) throws Exception {
    logger.info("onDuccJobsStateEvent", jobDeployment.getJobId(),
            "JP>>>>>>>>>>>> -PID:" + process.getPID());

    IDuccWorkExecutable dwe = (IDuccWorkExecutable) dw;
    boolean preemptable = true;
    preemptable = dw.getPreemptableStatus();

    // agent will check the state of JP
    // process and either start, stop,
    // or take no action
    agent.reconcileProcessStateAndTakeAction(lifecycleController, process, dwe.getCommandLine(),
            jobDeployment.getStandardInfo(), jobDeployment.getProcessMemoryAssignment(),
            jobDeployment.getJobId(), preemptable);

  }

  private void handleServiceState(IDuccJobDeployment jobDeployment, IDuccWork dw,
          IDuccProcess process) throws Exception {
    logger.info("onDuccJobsStateEvent", jobDeployment.getJobId(), "SERVICE>>>>>>>>>>>> -PID:"
            + process.getPID() + " Preemptable:" + dw.getPreemptableStatus());

    IDuccWorkJob service = (IDuccWorkJob) dw;
    ICommandLine processCmdLine = service.getCommandLine();
    processCmdLine.addOption("-Dducc.deploy.components=service");
    agent.reconcileProcessStateAndTakeAction(lifecycleController, process, processCmdLine,
            jobDeployment.getStandardInfo(), jobDeployment.getProcessMemoryAssignment(),
            jobDeployment.getJobId(), dw.getPreemptableStatus());

  }

  private void handleJobDriverState(IDuccJobDeployment jobDeployment,
          Map<DuccId, IDuccProcess> processes) throws Exception {
    IDuccWork dw = null;
    boolean callReconcile = false;
    boolean preemptable = false;
    ICommandLine cmdLine = null;
    // if driver not running, launch it
    if (!processes.containsKey(jobDeployment.getJdProcess().getDuccId())) {
      dw = dwHelper.fetch(jobDeployment.getJobId());
      if (dw == null) {
        logger.info("onDuccJobsStateEvent", jobDeployment.getJobId(),
                "!!!!! The OR did not provide commndline spec and other details required to launch processes for the Job. Received value of NULL from the OR");
      } else {
        IDuccWorkJob job = (IDuccWorkJob) dw;
        DuccWorkPopDriver driver = job.getDriver();
        if (driver != null) {
          callReconcile = true;
          cmdLine = driver.getCommandLine();
        }
        preemptable = dw.getPreemptableStatus();
      }
    } else {
      // driver running already. Still call reconcileProcessStateAndTakeAction()
      // in case we need to stop it if deallocate=true
      callReconcile = true;
    }
    if (callReconcile) {
      agent.reconcileProcessStateAndTakeAction(lifecycleController, jobDeployment.getJdProcess(),
              cmdLine, jobDeployment.getStandardInfo(), jobDeployment.getProcessMemoryAssignment(),
              jobDeployment.getJobId(), preemptable);

    }
  }

  private void handleAPJobServiceProcesses(IDuccJobDeployment jobDeployment,
          Map<DuccId, IDuccProcess> processes) throws Exception {
    // Service, AP, JP processes will be checked below
    if (jobDeployment.getJpProcessList() != null) {
      IDuccWork dw = null;
      for (IDuccProcess process : jobDeployment.getJpProcessList()) {
        // check if the process is targeted for this node
        if (isTargetNodeForProcess(process)) {
          // check if this is a new process we need to launch
          // on this node. The 'processes' is a Map of child
          // processes already running on this node.
          if (!processes.containsKey(process.getDuccId())) {
            // call the OR to fetch task details
            if ((dw = dwHelper.fetch(jobDeployment.getJobId())) == null) {
              logger.info("onDuccJobsStateEvent", jobDeployment.getJobId(),
                      "!!!!! The OR did not provide commndline spec and other details required to launch processes for the Job. Received value of NULL from the OR");
            } else {
              logger.debug("onDuccJobsStateEvent", jobDeployment.getJobId(),
                      "........... Job Type:" + jobDeployment.getType().name());

              if (DuccType.Service.equals(jobDeployment.getType())) {
                handleServiceState(jobDeployment, dw, process);
              } else if (dw instanceof IDuccWorkExecutable) {
                handleJobOrPOPState(jobDeployment, dw, process);
              } else {
                logger.error("onDuccJobsStateEvent", jobDeployment.getJobId(),
                        "Unexpected process type -PID:" + process.getPID());
              }
            }
          } else {
            // process already running, check state and stop it if deallocate=true
            agent.reconcileProcessStateAndTakeAction(lifecycleController, process, null,
                    jobDeployment.getStandardInfo(), jobDeployment.getProcessMemoryAssignment(),
                    jobDeployment.getJobId(), false);
          }
        }
      } // for
    }
  }

  /**
   * This method is called by Camel when PM sends DUCC state to agent's queue. It takes
   * responsibility of reconciling processes on this node.
   * 
   * @param duccEvent
   *          - contains list of current DUCC jobs.
   * @throws Exception
   */
  public void onDuccJobsStateEvent(@Body DuccJobsStateEvent duccEvent) throws Exception {
    String location = "onDuccJobsStateEvent";
    long sequence = duccEvent.getSequence();

    try {

      synchronized (this) {

        String host = duccEvent.getProducerHost();
        DuccHeadState dhs = duccEvent.getDuccHeadState();
        int jobs = 0;
        List<IDuccJobDeployment> listJobs = duccEvent.getJobList();
        if (listJobs != null) {
          jobs = listJobs.size();
        }
        int reservations = 0;
        List<DuccUserReservation> listReservations = duccEvent.getUserReservations();
        if (listReservations != null) {
          reservations = listReservations.size();
        }
        long tid = Thread.currentThread().getId();
        String message = "sequence=" + sequence + " " + "type=" + dhs + " " + "producer=" + host
                + " " + "jobs=" + jobs + " " + "reservations=" + reservations + " " + "tid=" + tid;
        switch (dhs) {
          case master:
          case unspecified:
            // Issue info and process master type publication
            logger.info(location, null, "accept=Yes" + " " + message);
            break;
          default:
            // Issue warning and ignore non-master type publication
            logger.warn(location, null, "accept=No" + " " + message);
            return;
        }

        // check for out of band messages. Expecting a message with a
        // sequence number larger than the previous message.
        if (sequence > lastSequence.get()) {
          lastSequence.set(sequence);
          logger.debug("reportIncomingStateForThisNode", null, "Received OR Sequence:" + sequence
                  + " Thread ID:" + Thread.currentThread().getId());
        } else {
          // Out of band message. Expected message with sequence
          // larger than a previous message
          logger.warn("reportIncomingStateForThisNode", null,
                  "Received Out of Band Message. Expected Sequence Greater Than " + lastSequence
                          + " Received " + sequence + " Instead");
          forceInventoryUpdateDueToSequence = true;
          return;
        }

        // typically lifecycleController is null and the agent assumes
        // the role. For jUnit testing though,
        // a different lifecycleController is injected to facilitate
        // black box testing
        if (lifecycleController == null) {
          lifecycleController = agent;
        }
        // print JP report targeted for this node
        reportIncomingStateForThisNode(duccEvent);

        agent.setReservations(duccEvent.getUserReservations());
        // Stop any process that is in this Agent's inventory but not
        // associated with any
        // of the jobs we just received
        agent.takeDownProcessWithNoJobV2(agent, duccEvent.getJobList());
        Map<DuccId, IDuccProcess> processes = agent.getInventoryCopy();
        // iterate over all jobs and reconcile those processes that are
        // assigned to this agent. First,
        // look at the job's JD process and than JPs.
        for (IDuccJobDeployment jobDeployment : duccEvent.getJobList()) {
          // check if this node is a target for this job's JD
          if (jobDeployment.getJdProcess() != null
                  && isTargetNodeForProcess(jobDeployment.getJdProcess())) {
            handleJobDriverState(jobDeployment, processes);
          }

          handleAPJobServiceProcesses(jobDeployment, processes);
        }
      }
      // received at least one Ducc State
      if (!agent.receivedDuccState) {
        agent.receivedDuccState = true;
      }
    } catch (Exception e) {
      logger.error("onDuccJobsStateEvent", null, e);
    }
  }

  /**
   * Wrapper method for Utils.isTargetNodeForMessage()
   * 
   * @param process
   * @return
   * @throws Exception
   */
  private boolean isTargetNodeForProcess(IDuccProcess process) throws Exception {
    boolean retVal = false;
    if (process != null) {
      retVal = Utils.isTargetNodeForMessage(process.getNodeIdentity().getIp(),
              agent.getIdentity().getIp());
    }
    return retVal;
  }

  public void onProcessStartEvent(@Body ProcessStartDuccEvent duccEvent) throws Exception {
    // iterate given ProcessMap and start a Process if this node is a target
    for (Entry<DuccId, IDuccProcess> processEntry : duccEvent.getProcessMap().entrySet()) {
      // check if this Process should be launched on this node. A process map
      // may contain processes not meant to run on this node. Each Process instance
      // in a Map has a node assignment. Only if this assignment matches this node
      // the agent will start a process.
      if (Utils.isTargetNodeForMessage(processEntry.getValue().getNodeIdentity().getIp(),
              agent.getIdentity().getIp())) {
        logger.info(">>> onProcessStartEvent", null,
                "... Agent [" + agent.getIdentity().getIp() + "] Matches Target Node Assignment:"
                        + processEntry.getValue().getNodeIdentity().getIp() + " For Share Id:"
                        + processEntry.getValue().getDuccId());
        agent.doStartProcess(processEntry.getValue(), duccEvent.getCommandLine(),
                duccEvent.getStandardInfo(), duccEvent.getDuccWorkId(), true);
        if (processEntry.getValue().getProcessType().equals(ProcessType.Pop)) {
          break; // there should only be one JD process to launch
        } else {
          continue;
        }
      }
    }
  }

  public void onProcessStopEvent(@Body ProcessStopDuccEvent duccEvent) throws Exception {
    for (Entry<DuccId, IDuccProcess> processEntry : duccEvent.getProcessMap().entrySet()) {
      if (Utils.isTargetNodeForMessage(processEntry.getValue().getNodeIdentity().getIp(),
              agent.getIdentity().getIp())) {
        logger.info(">>> onProcessStopEvent", null,
                "... Agent Received StopProces Event - Process Ducc Id:"
                        + processEntry.getValue().getDuccId() + " PID:"
                        + processEntry.getValue().getPID());
        agent.doStopProcess(processEntry.getValue());
      }
    }
  }

  public void onProcessStateUpdate(@Body ProcessStateUpdateDuccEvent duccEvent) throws Exception {
    logger.info(">>> onProcessStateUpdate", null,
            "... Agent Received ProcessStateUpdateDuccEvent - Process State:" + duccEvent.getState()
                    + " Process ID:" + duccEvent.getDuccProcessId());
    // agent.updateProcessStatus(duccEvent.getDuccProcessId(), duccEvent.getPid(),
    // duccEvent.getState());
    agent.updateProcessStatus(duccEvent);
  }

  public void onProcessStateUpdate(@Body String serviceUpdate) throws Exception {
    logger.info(">>> onProcessStateUpdate", null, "Recv'd Process Update >>> ");
    String[] stateUpdateProperties = serviceUpdate.split(",");

    String duccProcessId = null;
    String duccProcessState = null;
    String jmxUrl = null;
    ProcessState state = null;
    for (String prop : stateUpdateProperties) {
      String[] nv = prop.split("=");
      if (nv[0].equals("DUCC_PROCESS_UNIQUEID")) {
        duccProcessId = nv[1];
      } else if (nv[0].equals("DUCC_PROCESS_STATE")) {
        duccProcessState = nv[1];
        try {
          // validates value. Fails with IllegalArgumentException if
          // a value does not match a defined enum
          state = ProcessState.valueOf(nv[1]);
        } catch (IllegalArgumentException e) {
          // invalid state
        }
      } else if (nv[0].equals("SERVICE_JMX_PORT")) {
        jmxUrl = nv[1];
      }
    }
    if (state == null) {
      logger.info(">>> onProcessStateUpdate", null,
              "... Agent Received Invalid State Update event - Unsupported Process State:"
                      + duccProcessState + " Process ID:" + duccProcessId);
    } else if (duccProcessId == null) {
      logger.info(">>> onProcessStateUpdate", null,
              "... Agent Received Invalid State Update event - Process State:" + duccProcessState
                      + " Missing Process ID");
    } else {
      ProcessStateUpdate update = new ProcessStateUpdate(state, null, duccProcessId, jmxUrl);
      ProcessStateUpdateDuccEvent duccEvent = new ProcessStateUpdateDuccEvent(update);
      String jmxInfo = "";
      if (Objects.nonNull(jmxUrl)) {
        jmxInfo = " Service JMX connect url:" + jmxUrl;
      }
      logger.info(">>> onProcessStateUpdate", null,
              "... Agent Received ProcessStateUpdateDuccEvent - Process State:"
                      + duccEvent.getState() + " Process ID:" + duccEvent.getDuccProcessId()
                      + jmxInfo);
      agent.updateProcessStatus(duccEvent);
    }
  }

  public void onProcessPurgeEvent(@Body ProcessPurgeDuccEvent duccEvent) throws Exception {
    logger.info(">>> onProcessPurgeEvent", null, "... Agent Received ProcessPurgeDuccEvent -"
            + " Process ID:" + duccEvent.getProcess().getPID());
    agent.purgeProcess(duccEvent.getProcess());
  }

  public long getLastSequence() {
    return lastSequence.get();
  }
}