/*
 * 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.pm;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.camel.CamelContext;
import org.apache.uima.ducc.common.boot.DuccDaemonRuntimeProperties;
import org.apache.uima.ducc.common.boot.DuccDaemonRuntimeProperties.DaemonName;
import org.apache.uima.ducc.common.component.AbstractDuccComponent;
import org.apache.uima.ducc.common.main.DuccService;
import org.apache.uima.ducc.common.utils.DuccLogger;
import org.apache.uima.ducc.common.utils.id.DuccId;
import org.apache.uima.ducc.pm.helper.DuccWorkHelper;
import org.apache.uima.ducc.transport.cmdline.ICommandLine;
import org.apache.uima.ducc.transport.dispatcher.DuccEventDispatcher;
import org.apache.uima.ducc.transport.event.DuccEvent;
import org.apache.uima.ducc.transport.event.DuccJobsStateEvent;
import org.apache.uima.ducc.transport.event.PmStateDuccEvent;
import org.apache.uima.ducc.transport.event.common.DuccJobDeployment;
import org.apache.uima.ducc.transport.event.common.DuccUserReservation;
import org.apache.uima.ducc.transport.event.common.DuccWorkJob;
import org.apache.uima.ducc.transport.event.common.DuccWorkPop;
import org.apache.uima.ducc.transport.event.common.DuccWorkPopDriver;
import org.apache.uima.ducc.transport.event.common.DuccWorkReservation;
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.IDuccReservationMap;
import org.apache.uima.ducc.transport.event.common.IDuccUnits.MemoryUnits;
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.ProcessMemoryAssignment;

/**
 * The ProcessManager's main role is to receive Orchestrator updates, trim received state and
 * publish a new state to the agents. Trimming is done to remove state that is irrelevant to
 * the agents. 
 */
public class ProcessManagerComponent extends AbstractDuccComponent 
implements ProcessManager {
	private static final String[] jobHeaderArray=
		{"DuccId","ProcessType","PID","ProcessState","ResourceState","NodeIP","NodeName","DeallocationType","JMX Url"};
	private static final String jobHeaderFormat =
		"%1$-15s|%2$-20s|%3$-10s|%4$-15s|%5$-15s|%6$-13s|%7$-45s|%8$-25s|%9$-45s";
	
	private static String header;
	private static String tbl=String.format("%1$-158s"," ").replace(" ", "-");
	public static DuccLogger logger = new DuccLogger(ProcessManagerComponent.class, DuccComponent);
	
	private static DuccWorkHelper dwHelper = null;
	
	//	Dispatch component used to send messages to remote Agents
	private DuccEventDispatcher eventDispatcher;
  private int shareQuantum;
  private int fudgeFactor = 5; // default 5%
  
	public ProcessManagerComponent(CamelContext context, DuccEventDispatcher eventDispatcher) {
		super("ProcessManager",context);
		this.eventDispatcher = eventDispatcher;
		if ( System.getProperty("ducc.rm.share.quantum") != null && System.getProperty("ducc.rm.share.quantum").trim().length() > 0 ) {
	    shareQuantum = Integer.parseInt(System.getProperty("ducc.rm.share.quantum").trim());
		}
    if ( System.getProperty("ducc.agent.share.size.fudge.factor") != null &&
         System.getProperty("ducc.agent.share.size.fudge.factor").trim().length() > 0) {
      fudgeFactor = Integer.parseInt(System.getProperty("ducc.agent.share.size.fudge.factor").trim());
    }
		
		header = 
				String.format(jobHeaderFormat,jobHeaderArray[0],jobHeaderArray[1],jobHeaderArray[2],
						   jobHeaderArray[3],jobHeaderArray[4],jobHeaderArray[5],jobHeaderArray[6],
						   jobHeaderArray[7],jobHeaderArray[8]+"\n");
		dwHelper = new DuccWorkHelper();
	}
	public void start(DuccService service) throws Exception {
		super.start(service, null);
		DuccDaemonRuntimeProperties.getInstance().boot(DaemonName.ProcessManager,getProcessJmxUrl());
		logger.info("start", null, "--PM started - jmx URL:"+super.getProcessJmxUrl());
		
	}	
	public DuccLogger getLogger() {
	    return logger;
	  }
	
	/* New Code */
	
	private long normalizeMemory(String processMemoryAssignment, MemoryUnits units) {
		 //  Get user defined memory assignment for the JP
	    long normalizedProcessMemoryRequirements =
	            Long.parseLong(processMemoryAssignment);
	    // Normalize memory requirements for JPs into Gigs 
	    if ( units.equals(MemoryUnits.KB ) ) {
	      normalizedProcessMemoryRequirements = (int)normalizedProcessMemoryRequirements/(1024*1024);
	    } else if ( units.equals(MemoryUnits.MB ) ) {
	      normalizedProcessMemoryRequirements = (int)normalizedProcessMemoryRequirements/1024;
	    } else if ( units.equals(MemoryUnits.GB ) ) {
	      //  already normalized
	    } else if ( units.equals(MemoryUnits.TB ) ) {
	      normalizedProcessMemoryRequirements = (int)normalizedProcessMemoryRequirements*1024;
	    }
	    return normalizedProcessMemoryRequirements;
	}
	private int getShares(long normalizedProcessMemoryRequirements ) {
	    int shares = (int)normalizedProcessMemoryRequirements/shareQuantum;  // get number of shares
	    if ( (normalizedProcessMemoryRequirements % shareQuantum) > 0 ) shares++; // ciel
	    return shares;
	}

	private String getCmdLine(ICommandLine iCommandLine) {
		StringBuffer sb = new StringBuffer();
		if(iCommandLine != null) {
			String[] commandLine = iCommandLine.getCommandLine();
			if(commandLine != null) {
				for(String item : commandLine) {
					sb.append(item);
					sb.append(" ");
				}
			}
		}
		return sb.toString();
	}

	public void dispatchStateUpdateToAgents(Map<DuccId, IDuccWork> workMap, long sequence) {
    String methodName="dispatchStateUpdateToAgents";
	  try {
	    dumpState(workMap);
	    // Create a job list which the PM will dispatch to agents
	    List<IDuccJobDeployment> jobDeploymentList = new ArrayList<IDuccJobDeployment>();
	    List<DuccUserReservation> reservationList = new ArrayList<DuccUserReservation>();

	    //  Populate job list with job data that agents need. Don't copy data the agent doesnt
	    //  care about.
	    for( Entry<DuccId, IDuccWork> entry : workMap.entrySet() ) {
	      if ( entry.getValue() instanceof DuccWorkJob ) {
	        DuccWorkJob dcj = (DuccWorkJob)entry.getValue();
	        //  Create process list for each job
	        List<IDuccProcess> jobProcessList = new ArrayList<IDuccProcess>();
	        
	        long normalizedProcessMemoryRequirements = normalizeMemory(dcj.getSchedulingInfo().getShareMemorySize(),dcj.getSchedulingInfo().getShareMemoryUnits());
	        int shares = getShares(normalizedProcessMemoryRequirements);
	        long processAdjustedMemorySize = shares * shareQuantum * 1024;  
	        ProcessMemoryAssignment pma = new ProcessMemoryAssignment();
	        pma.setShares(shares);
	        pma.setNormalizedMemoryInMBs(processAdjustedMemorySize);
	        
	    
	        //  Copy job processes 
	        for( Entry<DuccId,IDuccProcess> jpProcess : dcj.getProcessMap().getMap().entrySet()) {
	          jobProcessList.add(jpProcess.getValue());
	        }
	        
	        
/*	        
	        if ( dcj.getUimaDeployableConfiguration() instanceof DuccUimaDeploymentDescriptor ) {
	          //  Add deployment UIMA AS deployment descriptor path
	          ((JavaCommandLine)dcj.getCommandLine()).
	            addArgument(((DuccUimaDeploymentDescriptor)dcj.getUimaDeployableConfiguration()).getDeploymentDescriptorPath());
	        }
*/	        
	        //  add fudge factor (5% default)  to adjust memory computed above 
	        processAdjustedMemorySize += (processAdjustedMemorySize * ((double)fudgeFactor/100));
	        pma.setMaxMemoryWithFudge(processAdjustedMemorySize);
	        
	        logger.debug(methodName,dcj.getDuccId(),"--------------- User Requested Memory For Process:"+dcj.getSchedulingInfo().getShareMemorySize()+dcj.getSchedulingInfo().getShareMemoryUnits()+" PM Calculated Memory Assignment of:"+processAdjustedMemorySize);
	        
	        ICommandLine driverCmdLine = null;
	        ICommandLine processCmdLine = null;
	        IDuccProcess driverProcess = null;
	        IDuccWork dw = null;
	        
	        switch(dcj.getDuccType()) {
	        case Job:
	          logger.debug(methodName, dcj.getDuccId(), "case: Job");
	          dw = dwHelper.fetch(dcj.getDuccId());
	          IDuccWorkJob job = (IDuccWorkJob) dw;
	          DuccWorkPopDriver driver = job.getDriver();
			  if(driver != null) {
				  driverCmdLine = driver.getCommandLine();
				  driverProcess = driver.getProcessMap().entrySet().iterator().next().getValue();
			  }
	          processCmdLine = job.getCommandLine();
	          break;
	        case Service:
	          logger.debug(methodName, dcj.getDuccId(), "case: Service");
	          dw = dwHelper.fetch(dcj.getDuccId());
	          IDuccWorkJob service = (IDuccWorkJob) dw;
	          processCmdLine = service.getCommandLine();
	          processCmdLine.addOption("-Dducc.deploy.components=service");
	          break;
	        default:
	          logger.debug(methodName, dcj.getDuccId(), "case: default");
	          dw = dwHelper.fetch(dcj.getDuccId());
	          if(dw instanceof IDuccWorkExecutable) {
	        	  IDuccWorkExecutable dwe = (IDuccWorkExecutable) dw;
	        	  processCmdLine = dwe.getCommandLine();
	          }
		      break;
	        }
	        
	        String dText = "n/a";
	        if(driverCmdLine != null) {
	        	dText = getCmdLine(driverCmdLine);
	        }
	        logger.trace(methodName, dcj.getDuccId(), "driver: "+dText);
	        
	        String pText = "n/a";
	        if(processCmdLine != null) {
	        	pText = getCmdLine(processCmdLine);
	        }
	        logger.trace(methodName, dcj.getDuccId(), "process: "+pText);
	        
	        jobDeploymentList.add( new DuccJobDeployment(dcj.getDuccId(), driverCmdLine,
	                           processCmdLine, 
	                           dcj.getStandardInfo(),
	                           driverProcess,
	                           pma,
	                           //processAdjustedMemorySize,
	                           jobProcessList ));
	      } else if (entry.getValue() instanceof DuccWorkReservation ) {
	        String userId = ((DuccWorkReservation) entry.getValue()).getStandardInfo().getUser();
	        if ( !"System".equals(userId)) {
	          IDuccReservationMap reservationMap = 
	                  ((DuccWorkReservation) entry.getValue()).getReservationMap();
	          reservationList.add(new DuccUserReservation(userId, ((DuccWorkReservation) entry.getValue()).getDuccId(), reservationMap));
	          logger.debug(methodName,null,"---------------  Added reservation for user:"+userId);
	        }
        }
	  }
      logger.info(methodName, null , "---- PM Dispatching DuccJobsStateEvent request to Agent(s) - State Map Size:"+jobDeploymentList.size()+" Reservation List:"+reservationList.size());
      DuccJobsStateEvent ev =  new DuccJobsStateEvent(DuccEvent.EventType.PM_STATE, jobDeploymentList, reservationList);
      ev.setSequence(sequence);
      //  Dispatch state update to agents
      eventDispatcher.dispatch(ev);
      logger.debug(methodName, null , "+++++ PM Dispatched State To Agent(s)");
	  } catch( Throwable t ) {
      logger.error(methodName,null,t);
	  }
	}
	
	private String formatProcess( IDuccProcess process ) {
		return String.format(jobHeaderFormat,
			    String.valueOf(process.getDuccId().getFriendly()),
				process.getProcessType().toString(),
				(process.getPID()==null? "" :process.getPID()),
				process.getProcessState().toString(),
				process.getResourceState().toString(),
				process.getNodeIdentity().getIp(),
				process.getNodeIdentity().getName(),
				process.getProcessDeallocationType().toString(),
				(process.getProcessJmxUrl() == null ? "N/A" : process.getProcessJmxUrl() ));
		
	}
	private void dumpState(Map<DuccId, IDuccWork> workMap) {
		String methodName="dumpState";
		try {
			StringBuffer sb = new StringBuffer();
			for( Entry<DuccId,IDuccWork> job : workMap.entrySet()) {
				IDuccWork duccWork = job.getValue();
				if ( duccWork instanceof DuccWorkJob ) {
					DuccWorkJob duccWorkJob = (DuccWorkJob)duccWork;
					sb.append("\n").append(tbl).
					   append("\nJob ID: ").append(duccWorkJob.getDuccId().getFriendly()).
					   append("\tJobState: ").append(duccWorkJob.getStateObject()).
					   append("\tJobSubmittedBy: ").append(duccWorkJob.getStandardInfo().getUser()).
					   append("\n\n").
					   append(header).append(tbl).append("\n");
					DuccWorkPopDriver driver = duccWorkJob.getDriver();
					if(driver != null) {
						IDuccProcess driverProcess =
							driver.getProcessMap().entrySet().iterator().next().getValue();
							sb.append(formatProcess(driverProcess));
					}
					for(Entry<DuccId,IDuccProcess> process : ((DuccWorkJob)job.getValue()).getProcessMap().entrySet()) {
						sb.append("\n").append(formatProcess(process.getValue()));
					}
					sb.append("\n").append(tbl).append("\n");
					logger.info(methodName, null, sb.toString());
				} else if ( job.getValue() instanceof DuccWorkReservation ) {
					continue;  // TBI
				} else if ( job.getValue() instanceof DuccWorkPop ) {
					continue;  // TBI
				} else {
					logger.info(methodName, job.getKey(), "Not a WorkJob but "+job.getClass().getName());
				}
			}
			
		} catch( Exception e) {
			e.printStackTrace();
		}
	} 
	/**
	 * Override
	 */
	public void setLogLevel(String clz, String level) {
		logger.info("setLogLevel",null,"--------- Changing Log Level to:"+level+ " For class:"+clz);
		super.setLogLevel(clz, level);
	}
	public void setLogLevel(String level) {
		logger.info("setLogLevel",null,"--------- Changing Log Level to:"+level+ " For class:"+getClass().getCanonicalName());
		super.setLogLevel(getClass().getCanonicalName(), level);
	}
	public String getLogLevel() {
		return super.getLogLevel();
	}
	public void logAtTraceLevel(String toLog, String methodName) {
		if ( logger.isTrace()) {
			logger.trace(methodName,null,"--------- "+ toLog);
		}
	}
	

	public PmStateDuccEvent getState() {
		String methodName = "PmStateDuccEvent";
		logger.trace(methodName,null,"");
		return new PmStateDuccEvent();
	}

}