content: /**
   * 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
   * <p>
   * http://www.apache.org/licenses/LICENSE-2.0
   * <p>
   * 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.oozie.ambari.view;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  
  import javax.ws.rs.HttpMethod;
  import javax.ws.rs.core.HttpHeaders;
  import javax.ws.rs.core.MediaType;
  import javax.ws.rs.core.MultivaluedMap;
  import javax.ws.rs.core.Response;
  
  import org.apache.ambari.view.ViewContext;
  import org.apache.commons.io.IOUtils;
  import org.apache.oozie.ambari.view.exception.ErrorCode;
  import org.apache.oozie.ambari.view.exception.WfmException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class OozieDelegate {
    private final static Logger LOGGER = LoggerFactory
      .getLogger(OozieDelegate.class);
    private static final String OOZIEPARAM_PREFIX = "oozieparam.";
    private static final int OOZIEPARAM_PREFIX_LENGTH = OOZIEPARAM_PREFIX
      .length();
    private static final String EQUAL_SYMBOL = "=";
    private static final String OOZIE_WF_RERUN_FAILNODES_CONF_KEY = "oozie.wf.rerun.failnodes";
    private static final String OOZIE_USE_SYSTEM_LIBPATH_CONF_KEY = "oozie.use.system.libpath";
    private static final String USER_NAME_HEADER = "user.name";
    private static final String USER_OOZIE_SUPER = "oozie";
    private static final String DO_AS_HEADER = "doAs";
    private static final String SERVICE_URI_PROP = "oozie.service.uri";
    private static final String DEFAULT_SERVICE_URI = "http://sandbox.hortonworks.com:11000/oozie";
  
    private ViewContext viewContext;
  
    private OozieUtils oozieUtils = new OozieUtils();
    private final Utils utils = new Utils();
    private final AmbariIOUtil ambariIOUtil;
  
    public OozieDelegate(ViewContext viewContext) {
      super();
      this.viewContext = viewContext;
      this.ambariIOUtil = new AmbariIOUtil(viewContext);
    }
  
    public String submitWorkflowJobToOozie(HttpHeaders headers,
                                           String filePath, MultivaluedMap<String, String> queryParams,
                                           JobType jobType) {
  	String nameNode = viewContext.getProperties().get("webhdfs.url");
  	if (nameNode == null) {
  		LOGGER.error("Name Node couldn't be determined automatically.");
  		throw new RuntimeException("Name Node couldn't be determined automatically.");
  	}
  
      if (!queryParams.containsKey("config.nameNode")) {
        ArrayList<String> nameNodes = new ArrayList<String>();
        LOGGER.info("Namenode===" + nameNode);
        nameNodes.add(nameNode);
        queryParams.put("config.nameNode", nameNodes);
      }
  
      Map<String, String> workflowConigs = getWorkflowConfigs(filePath,
        queryParams, jobType, nameNode);
      String configXMl = oozieUtils.generateConfigXml(workflowConigs);
      LOGGER.info("Config xml==" + configXMl);
      HashMap<String, String> customHeaders = new HashMap<String, String>();
      customHeaders.put("Content-Type", "application/xml;charset=UTF-8");
      Response serviceResponse = consumeService(headers, getServiceUri()
          + "/v2/jobs?" + getJobSumbitOozieParams(queryParams),
        HttpMethod.POST, configXMl, customHeaders);
  
      LOGGER.info("Resp from oozie status entity=="
        + serviceResponse.getEntity());
      String oozieResp=null;
      if (serviceResponse.getEntity() instanceof String) {
        oozieResp= (String) serviceResponse.getEntity();
      } else {
        oozieResp= serviceResponse.getEntity().toString();
      }
      if (oozieResp != null && oozieResp.trim().startsWith("{")) {
        return  oozieResp;
      }else{
        throw new WfmException(oozieResp,ErrorCode.OOZIE_SUBMIT_ERROR);
      }
    }
  
    public Response consumeService(HttpHeaders headers, String path,
                                   MultivaluedMap<String, String> queryParameters, String method,
                                   String body) throws Exception {
      return consumeService(headers, this.buildUri(path, queryParameters),
        method, body, null);
    }
  
    private Response consumeService(HttpHeaders headers, String urlToRead,
                                    String method, String body, Map<String, String> customHeaders) {
      Response response = null;
      InputStream stream = readFromOozie(headers, urlToRead, method, body,
        customHeaders);
      String stringResponse = null;
      try {
        stringResponse = IOUtils.toString(stream);
      } catch (IOException e) {
        LOGGER.error("Error while converting stream to string", e);
        throw new RuntimeException(e);
      }
      if (stringResponse.contains(Response.Status.BAD_REQUEST.name())) {
        response = Response.status(Response.Status.BAD_REQUEST)
          .entity(stringResponse).type(MediaType.TEXT_PLAIN).build();
      } else {
        response = Response.status(Response.Status.OK)
          .entity(stringResponse)
          .type(utils.deduceType(stringResponse)).build();
      }
      return response;
    }
  
    public InputStream readFromOozie(HttpHeaders headers, String urlToRead,
                                     String method, String body, Map<String, String> customHeaders) {
  
      Map<String, String> newHeaders = utils.getHeaders(headers);
      newHeaders.put(USER_NAME_HEADER, USER_OOZIE_SUPER);
  
      newHeaders.put(DO_AS_HEADER, viewContext.getUsername());
      newHeaders.put("Accept", MediaType.APPLICATION_JSON);
      if (customHeaders != null) {
        newHeaders.putAll(customHeaders);
      }
      LOGGER.info(String.format("Proxy request for url: [%s] %s", method,
        urlToRead));
  
      return ambariIOUtil.readFromUrl(urlToRead, method, body, newHeaders);
    }
  
    private Map<String, String> getWorkflowConfigs(String filePath,
                                                   MultivaluedMap<String, String> queryParams, JobType jobType,
                                                   String nameNode) {
      HashMap<String, String> workflowConigs = new HashMap<String, String>();
      if (queryParams.containsKey("resourceManager")
        && "useDefault".equals(queryParams.getFirst("resourceManager"))) {
  	  String jobTrackerNode = viewContext.getProperties()
  	      .get("yarn.resourcemanager.address");
        LOGGER.info("jobTrackerNode===" + jobTrackerNode);
        workflowConigs.put("resourceManager", jobTrackerNode);
        workflowConigs.put("jobTracker", jobTrackerNode);
      }
      if (queryParams != null) {
        for (Map.Entry<String, List<String>> entry : queryParams.entrySet()) {
          if (entry.getKey().startsWith("config.")) {
            if (entry.getValue() != null && entry.getValue().size() > 0) {
              workflowConigs.put(entry.getKey().substring(7), entry
                .getValue().get(0));
            }
          }
        }
      }
  
      if (queryParams.containsKey("oozieconfig.useSystemLibPath")) {
        String useSystemLibPath = queryParams
          .getFirst("oozieconfig.useSystemLibPath");
        workflowConigs.put(OOZIE_USE_SYSTEM_LIBPATH_CONF_KEY,
          useSystemLibPath);
      } else {
        workflowConigs.put(OOZIE_USE_SYSTEM_LIBPATH_CONF_KEY, "true");
      }
      if (queryParams.containsKey("oozieconfig.rerunOnFailure")) {
        String rerunFailnodes = queryParams
          .getFirst("oozieconfig.rerunOnFailure");
        workflowConigs.put(OOZIE_WF_RERUN_FAILNODES_CONF_KEY,
          rerunFailnodes);
      } else {
        workflowConigs.put(OOZIE_WF_RERUN_FAILNODES_CONF_KEY, "true");
      }
      workflowConigs.put("user.name", viewContext.getUsername());
      workflowConigs.put(oozieUtils.getJobPathPropertyKey(jobType), nameNode
        + filePath);
      return workflowConigs;
    }
  
    private String getJobSumbitOozieParams(
      MultivaluedMap<String, String> queryParams) {
      StringBuilder query = new StringBuilder();
      if (queryParams != null) {
        for (Map.Entry<String, List<String>> entry : queryParams.entrySet()) {
          if (entry.getKey().startsWith(OOZIEPARAM_PREFIX)) {
            if (entry.getValue() != null && entry.getValue().size() > 0) {
              for (String val : entry.getValue()) {
                query.append(
                  entry.getKey().substring(
                    OOZIEPARAM_PREFIX_LENGTH))
                  .append(EQUAL_SYMBOL).append(val)
                  .append("&");
              }
            }
          }
        }
      }
      return query.toString();
    }
  
    private String getServiceUri() {
      String serviceURI = viewContext.getProperties().get(SERVICE_URI_PROP) != null ? viewContext
        .getProperties().get(SERVICE_URI_PROP) : DEFAULT_SERVICE_URI;
      return serviceURI;
    }
  
    private String buildUri(String absolutePath,
                            MultivaluedMap<String, String> queryParameters) {
      int index = absolutePath.indexOf("proxy/") + 5;
      absolutePath = absolutePath.substring(index);
      String serviceURI = getServiceUri();
      serviceURI += absolutePath;
      MultivaluedMap<String, String> params = addOrReplaceUserName(queryParameters);
      return serviceURI + utils.convertParamsToUrl(params);
    }
  
    private MultivaluedMap<String, String> addOrReplaceUserName(
      MultivaluedMap<String, String> parameters) {
      for (Map.Entry<String, List<String>> entry : parameters.entrySet()) {
        if ("user.name".equals(entry.getKey())) {
          ArrayList<String> vals = new ArrayList<String>(1);
          vals.add(viewContext.getUsername());
          entry.setValue(vals);
        }
      }
      return parameters;
    }
  
    public String getDagUrl(String jobid) {
      return getServiceUri() + "/v2/job/" + jobid + "?show=graph";
    }
  }
  