// Copyright 2016 Twitter. All rights reserved.
//
// 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 com.twitter.heron.scheduler.aurora;

import java.net.HttpURLConnection;
import java.util.Map;
import java.util.Set;
import java.util.logging.Logger;

import com.twitter.heron.proto.system.PhysicalPlans.StMgr;
import com.twitter.heron.spi.common.Config;
import com.twitter.heron.spi.common.Context;
import com.twitter.heron.spi.packing.PackingPlan;
import com.twitter.heron.spi.statemgr.IStateManager;
import com.twitter.heron.spi.statemgr.SchedulerStateManagerAdaptor;
import com.twitter.heron.spi.utils.NetworkUtils;
import com.twitter.heron.spi.utils.ReflectionUtils;

/**
 * Implementation of AuroraController that is a wrapper of AuroraCLIController The difference is:
 * the `restart` method implementation is changed to heron-shell
 */
class AuroraHeronShellController implements AuroraController {
  private static final Logger LOG = Logger.getLogger(AuroraHeronShellController.class.getName());

  private final String topologyName;
  private final AuroraCLIController cliController;
  private final SchedulerStateManagerAdaptor stateMgrAdaptor;

  AuroraHeronShellController(String jobName, String cluster, String role, String env,
      String auroraFilename, boolean isVerbose, Config localConfig)
      throws ClassNotFoundException, InstantiationException, IllegalAccessException {
    this.topologyName = jobName;
    this.cliController =
        new AuroraCLIController(jobName, cluster, role, env, auroraFilename, isVerbose);

    Config config = Config.toClusterMode(localConfig);
    String stateMgrClass = Context.stateManagerClass(config);
    IStateManager stateMgr = ReflectionUtils.newInstance(stateMgrClass);
    stateMgr.initialize(config);
    stateMgrAdaptor = new SchedulerStateManagerAdaptor(stateMgr, 5000);
  }

  @Override
  public boolean createJob(Map<AuroraField, String> bindings) {
    return cliController.createJob(bindings);
  }

  @Override
  public boolean killJob() {
    return cliController.killJob();
  }

  // Restart an aurora container
  @Override
  public boolean restart(Integer containerId) {
    if (containerId == null) {
      throw new UnsupportedOperationException("Not implemented");
    }

    if (stateMgrAdaptor == null) {
      LOG.warning("SchedulerStateManagerAdaptor not initialized");
      return false;
    }

    String url = null;
    if (containerId == 0) {
      // TODO(huijun): needs shell port for container 0
      throw new UnsupportedOperationException("Not implemented for container 0");
    } else {
      int index = containerId - 1; // stmgr container starts from 1
      StMgr containerInfo = stateMgrAdaptor.getPhysicalPlan(topologyName).getStmgrs(index);
      String host = containerInfo.getHostName();
      int port = containerInfo.getShellPort();
      url = "http://" + host + ":" + port + "/killexecutor";
    }

    String payload = "secret=" + stateMgrAdaptor.getExecutionState(topologyName).getTopologyId();
    LOG.info("sending `kill container` to " + url + "; payload: " + payload);

    HttpURLConnection con = NetworkUtils.getHttpConnection(url);
    try {
      NetworkUtils.sendHttpPostRequest(con, "X", payload.getBytes());
      return NetworkUtils.checkHttpResponseCode(con, 200);
    } finally {
      con.disconnect();
    }
  }

  @Override
  public void removeContainers(Set<PackingPlan.ContainerPlan> containersToRemove) {
    cliController.removeContainers(containersToRemove);
  }

  @Override
  public void addContainers(Integer count) {
    cliController.addContainers(count);
  }
}

@@ -1257,16 +1257,20 @@
 ntroller
+.%0A *
  The dif
@@ -1283,74 +1283,454 @@
 e is
-:%0A * the %60restart%60 method implementation is changed to heron-shell
+ %60restart%60 command:%0A * 1. restart whole topology: delegate to AuroraCLIController%0A * 2. restart container 0: delegate to AuroraCLIController%0A * 3. restart container x(x%3E0): call heron-shell endpoint %60/killexecutor%60%0A * For backpressure, only containers with heron-stmgr may send out backpressure.%0A * This class is to handle %60restart backpressure containers inside container%60,%0A * while delegating to AuroraCLIController for all the other scenarios.
 %0A */
@@ -3041,441 +3041,308 @@
     
-if (containerId == null) %7B%0A      throw new UnsupportedOperationException(%22Not implemented%22);%0A    %7D%0A%0A    if (stateMgrAdaptor == null) %7B%0A      LOG.warning(%22SchedulerStateManagerAdaptor not initialized%22);%0A      return false;%0A    %7D%0A%0A    String url = null;%0A    if (containerId == 0) %7B%0A      // TODO(huijun): needs shell port for container 0%0A      throw new UnsupportedOperationException(%22Not implemented for container 0%22);%0A    %7D else %7B%0A  
+// there is no backpressure for container 0, delegate to aurora client%0A    if (containerId == null %7C%7C containerId == 0) %7B%0A      cliController.restart(containerId);%0A    %7D%0A%0A    if (stateMgrAdaptor == null) %7B%0A      LOG.warning(%22SchedulerStateManagerAdaptor not initialized%22);%0A      return false;%0A    %7D%0A%0A
     
@@ -3407,18 +3407,16 @@
 m 1%0A    
-  
 StMgr co
@@ -3415,26 +3415,26 @@
 StMgr contai
-n
 e
+n
 rInfo = stat
@@ -3493,26 +3493,24 @@
 index);%0A    
-  
 String host 
@@ -3509,34 +3509,34 @@
 ng host = contai
-n
 e
+n
 rInfo.getHostNam
@@ -3544,18 +3544,16 @@
 ();%0A    
- 
- 
 int port
@@ -3561,18 +3561,18 @@
 = contai
-n
 e
+n
 rInfo.ge
@@ -3589,17 +3589,22 @@
 ();%0A    
- 
+String
  url = %22
@@ -3651,23 +3651,16 @@
 cutor%22;%0A
-    %7D%0A%0A
     Stri

update auroraheronshellcontroller (#2419)