This project has retired. For details please refer to its Attic page.
ExecPlugin xref
View Javadoc

1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  
19  package org.apache.hadoop.chukwa.inputtools.plugin;
20  
21  
22  import java.io.BufferedReader;
23  import java.io.IOException;
24  import java.io.InputStream;
25  import java.io.InputStreamReader;
26  import java.nio.charset.Charset;
27  
28  import org.json.simple.JSONObject;
29  
30  /**
31   * Runs external command-line tools, captures output.
32   * 
33   * Subclasses are responsible for implementing getCmde(), which determines the
34   * command to be invoked.
35   * 
36   */
37  public abstract class ExecPlugin implements IPlugin {
38    public final static int statusOK = 100;
39    public final static int statusKO = -100;
40  
41    Process process = null;
42  
43    public ExecPlugin() {
44  
45    }
46  
47    public void stop() {
48      if(process != null)
49        process.destroy();
50    }
51  
52    public int waitFor() throws InterruptedException {
53      return process.waitFor();
54    }
55  
56    public abstract String getCmde();
57  
58    public JSONObject postProcess(JSONObject execResult) {
59      return execResult;
60    }
61  
62    public JSONObject execute() {
63      JSONObject result = new JSONObject();
64      try {
65        result.put("timestamp", System.currentTimeMillis());
66  
67        Runtime runtime = Runtime.getRuntime();
68        process = runtime.exec(getCmde());
69  
70        OutputReader stdOut = new OutputReader(process, Output.stdOut);
71        stdOut.start();
72        OutputReader stdErr = new OutputReader(process, Output.stdErr);
73        stdErr.start();
74        int exitValue = process.waitFor();
75        stdOut.join();
76        stdErr.join();
77        process.getInputStream().close(); //otherwise this implicitly stays open
78        result.put("exitValue", exitValue);
79        result.put("stdout", stdOut.output.toString());
80        result.put("stderr", stdErr.output.toString());
81        result.put("status", statusOK);
82        process.getOutputStream().close();
83        process.getErrorStream().close();
84      } catch (Throwable e) {
85        try {
86          result.put("status", statusKO);
87          result.put("errorLog", e.getMessage());
88          if(e.getMessage().contains("Too many open files")) {
89            //maybe die abruptly?  Error is ir-recoverable and runtime can reboot us.
90  //        System.exit(1); 
91          }
92        } catch (Exception e1) {
93          e1.printStackTrace();
94        }
95        e.printStackTrace();
96      }
97  
98      return postProcess(result);
99    }
100 }
101 
102 
103 enum Output {
104   stdOut, stdErr
105 };
106 
107 
108 class OutputReader extends Thread {
109   private Process process = null;
110   private Output outputType = null;
111   public StringBuilder output = new StringBuilder();
112 
113   public OutputReader(Process process, Output outputType) {
114     this.process = process;
115     this.outputType = outputType;
116   }
117 
118   public void run() {
119     try {
120       String line = null;
121       InputStream is = null;
122       switch (this.outputType) {
123       case stdOut:
124         is = process.getInputStream();
125         break;
126       case stdErr:
127         is = process.getErrorStream();
128         break;
129 
130       }
131       if(is!=null) {
132         InputStreamReader isr = new InputStreamReader(is, Charset.forName("UTF-8"));
133         BufferedReader br = new BufferedReader(isr);
134         while ((line = br.readLine()) != null) {
135           // System.out.println("========>>>>>>>["+line+"]");
136           output.append(line).append("\n");
137         }
138         br.close();
139       }
140     } catch (IOException e) {
141       e.printStackTrace();
142     } catch (Throwable e) {
143       e.printStackTrace();
144     }
145   }
146 }