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
   *
   *     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.hadoop.chukwa.datacollection.adaptor;
  
  import java.io.FileInputStream;
  import java.nio.charset.Charset;
  import java.security.KeyStore;
  import java.security.SecureRandom;
  import java.util.Calendar;
  import java.util.TimeZone;
  import java.util.Timer;
  import java.util.TimerTask;
  
  import org.apache.hadoop.chukwa.ChunkImpl;
  import org.apache.hadoop.chukwa.datacollection.ChunkReceiver;
  import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
  import org.apache.log4j.Logger;
  import org.apache.hadoop.chukwa.util.ExceptionUtil;
  import org.apache.hadoop.conf.Configuration;
  
  import static org.apache.hadoop.chukwa.datacollection.agent.ChukwaConstants.*;
  
  import com.sun.jersey.api.client.Client;
  import com.sun.jersey.api.client.WebResource;
  import com.sun.jersey.api.client.config.ClientConfig;
  import com.sun.jersey.api.client.config.DefaultClientConfig;
  import com.sun.jersey.client.urlconnection.HTTPSProperties;
  
  import javax.net.ssl.SSLContext;
  import javax.net.ssl.TrustManager;
  import javax.net.ssl.TrustManagerFactory;
  import javax.ws.rs.core.MediaType;
  
  public class RestAdaptor extends AbstractAdaptor {
  
    private String uri;
    private long period = 60;
    private static Logger log = Logger.getLogger(RestAdaptor.class);
    private WebResource resource;
    private Client c;
    private String bean;
    private Timer timer;
    private TimerTask runner;
    private long sendOffset;
  
    class RestTimer extends TimerTask {
  
      private ChunkReceiver receiver;
      private RestAdaptor adaptor;
  
      RestTimer(ChunkReceiver receiver, RestAdaptor adaptor) {
        this.receiver = receiver;
        this.adaptor = adaptor;
      }
  
      @Override
      public void run() {
        try {
          resource = c.resource(uri);
          bean = resource.accept(MediaType.APPLICATION_JSON_TYPE).get(
              String.class);
          byte[] data = bean.getBytes(Charset.forName("UTF-8"));
          sendOffset += data.length;
          ChunkImpl c = new ChunkImpl(type, "REST", sendOffset, data, adaptor);
          long rightNow = Calendar.getInstance(TimeZone.getTimeZone("UTC"))
              .getTimeInMillis();
          c.addTag("timeStamp=\"" + rightNow + "\"");
          receiver.add(c);
        } catch (com.sun.jersey.api.client.ClientHandlerException e) {
          Throwable t = e.getCause();
          if (t instanceof java.net.ConnectException) {
            log.warn("Connect exception trying to connect to " + uri
                + ". Make sure the service is running");
          } else {
            log.error("RestAdaptor: Interrupted exception");
            log.error(ExceptionUtil.getStackTrace(e));
          }
        } catch (Exception e) {
          log.error("RestAdaptor: Interrupted exception");
          log.error(ExceptionUtil.getStackTrace(e));
        }
      }
    }
  
    @Override
    public String getCurrentStatus() {
      StringBuilder buffer = new StringBuilder();
      buffer.append(type);
      buffer.append(" ");
      buffer.append(uri);
      buffer.append(" ");
      buffer.append(period);
      return buffer.toString();
    }
  
    @Override
    public long shutdown(AdaptorShutdownPolicy shutdownPolicy)
        throws AdaptorException {
      timer.cancel();
      return sendOffset;
    }
  
    @Override
    public void start(long offset) throws AdaptorException {
      sendOffset = offset;
      if (timer == null) {
        timer = new Timer();
        runner = new RestTimer(dest, RestAdaptor.this);
      }
      timer.scheduleAtFixedRate(runner, 0, period * 1000);
    }
  
    @Override
    public String parseArgs(String s) {
      // RestAdaptor [Host] port uri [interval]
      String[] tokens = s.split(" ");
      if (tokens.length == 2) {
        uri = tokens[0];
        try {
          period = Integer.parseInt(tokens[1]);
        } catch (NumberFormatException e) {
          log.warn("RestAdaptor: incorrect argument for period. Expecting number");
          return null;
        }
      } else {
        log.warn("bad syntax in RestAdaptor args");
        return null;
      }
      try {
        initClient();
      } catch (Exception e) {
        log.error(ExceptionUtil.getStackTrace(e));
        return null;
      }    
      return s;
    }
    
    private void initClient() throws Exception {
      if (uri.contains("https")) {
        Configuration conf = ChukwaAgent.getAgent().getConfiguration();
        String trustStoreFile = conf.get(TRUSTSTORE_STORE);
        String trustStorePw = conf.get(TRUST_PASSWORD);
        if (trustStoreFile == null || trustStorePw == null) {
          throw new Exception(
              "Cannot instantiate RestAdaptor to uri "
                  + uri
                  + " due to missing trust store configurations chukwa.ssl.truststore.store and chukwa.ssl.trust.password");
        }
        String trustStoreType = conf.get(TRUSTSTORE_TYPE, DEFAULT_STORE_TYPE);
        KeyStore trustStore = KeyStore.getInstance(trustStoreType);
        FileInputStream fis = null;
        try {
          fis = new FileInputStream(trustStoreFile);
          trustStore.load(fis, trustStorePw.toCharArray());
        } finally {
          if (fis != null) {
            fis.close();
          }
        }
        TrustManagerFactory tmf = TrustManagerFactory
            .getInstance(TrustManagerFactory.getDefaultAlgorithm());
        tmf.init(trustStore);
        TrustManager[] trustManagers = tmf.getTrustManagers();
  
        SSLContext ctx = null;
        String protocol = conf.get(SSL_PROTOCOL, DEFAULT_SSL_PROTOCOL);
        ctx = SSLContext.getInstance(protocol);
        ctx.init(null, trustManagers, new SecureRandom());
        ClientConfig cc = new DefaultClientConfig();
        HTTPSProperties props = new HTTPSProperties(null, ctx);
        cc.getProperties().put(HTTPSProperties.PROPERTY_HTTPS_PROPERTIES, props);
        c = Client.create(cc);
      } else {
        c = Client.create();
      }
    }
  
  }
  