View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.monitoring;
20  
21  import org.apache.hadoop.hbase.classification.InterfaceAudience;
22  import org.apache.hadoop.util.StringUtils;
23  import org.codehaus.jackson.map.ObjectMapper;
24  
25  import java.io.IOException;
26  import java.util.ArrayList;
27  import java.util.Collections;
28  import java.util.HashMap;
29  import java.util.List;
30  import java.util.Map;
31  
32  @InterfaceAudience.Private
33  class MonitoredTaskImpl implements MonitoredTask {
34    private long startTime;
35    private long statusTime;
36    private long stateTime;
37    private long warnTime;
38  
39    private volatile String status;
40    private volatile String description;
41    
42    protected volatile State state = State.RUNNING;
43  
44    private boolean journalEnabled = false;
45    private List<StatusJournalEntry> journal;
46  
47    private static final ObjectMapper MAPPER = new ObjectMapper();
48  
49    public MonitoredTaskImpl() {
50      startTime = System.currentTimeMillis();
51      statusTime = startTime;
52      stateTime = startTime;
53      warnTime = startTime;
54    }
55  
56    private static class StatusJournalEntryImpl implements StatusJournalEntry {
57      private long statusTime;
58      private String status;
59  
60      public StatusJournalEntryImpl(String status, long statusTime) {
61        this.status = status;
62        this.statusTime = statusTime;
63      }
64  
65      @Override
66      public String getStatus() {
67        return status;
68      }
69  
70      @Override
71      public long getTimeStamp() {
72        return statusTime;
73      }
74  
75      @Override
76      public String toString() {
77        StringBuilder sb = new StringBuilder();
78        sb.append(status);
79        sb.append(" at ");
80        sb.append(statusTime);
81        return sb.toString();
82      }
83    }
84  
85    @Override
86    public synchronized MonitoredTaskImpl clone() {
87      try {
88        return (MonitoredTaskImpl) super.clone();
89      } catch (CloneNotSupportedException e) {
90        throw new AssertionError(); // Won't happen
91      }
92    }
93  
94    @Override
95    public long getStartTime() {
96      return startTime;
97    }
98    
99    @Override
100   public String getDescription() {
101     return description;
102   }
103 
104   @Override
105   public String getStatus() {
106     return status;
107   }
108 
109   @Override
110   public long getStatusTime() {
111     return statusTime;
112   }
113   
114   @Override
115   public State getState() {
116     return state;
117   }
118   
119   @Override
120   public long getStateTime() {
121     return stateTime;
122   }
123 
124   @Override
125   public long getWarnTime() {
126     return warnTime;
127   }
128 
129   @Override
130   public long getCompletionTimestamp() {
131     if (state == State.COMPLETE || state == State.ABORTED) {
132       return stateTime;
133     }
134     return -1;
135   }
136 
137   @Override
138   public void markComplete(String status) {
139     setState(State.COMPLETE);
140     setStatus(status);
141   }
142 
143   @Override
144   public void pause(String msg) {
145     setState(State.WAITING);
146     setStatus(msg);
147   }
148 
149   @Override
150   public void resume(String msg) {
151     setState(State.RUNNING);
152     setStatus(msg);
153   }
154 
155   @Override
156   public void abort(String msg) {
157     setStatus(msg);
158     setState(State.ABORTED);
159   }
160   
161   @Override
162   public void setStatus(String status) {
163     this.status = status;
164     statusTime = System.currentTimeMillis();
165     if (journalEnabled) {
166       journal.add(new StatusJournalEntryImpl(this.status, statusTime));
167     }
168   }
169 
170   protected void setState(State state) {
171     this.state = state;
172     stateTime = System.currentTimeMillis();
173   }
174 
175   @Override
176   public void setDescription(String description) {
177     this.description = description;
178   }
179 
180   @Override
181   public void setWarnTime(long t) {
182     this.warnTime = t;
183   }
184 
185   @Override
186   public void cleanup() {
187     if (state == State.RUNNING) {
188       setState(State.ABORTED);
189     }
190   }
191 
192   /**
193    * Force the completion timestamp backwards so that
194    * it expires now.
195    */
196   public void expireNow() {
197     stateTime -= 180 * 1000;
198   }
199 
200   @Override
201   public Map<String, Object> toMap() {
202     Map<String, Object> map = new HashMap<String, Object>();
203     map.put("description", getDescription());
204     map.put("status", getStatus());
205     map.put("state", getState());
206     map.put("starttimems", getStartTime());
207     map.put("statustimems", getCompletionTimestamp());
208     map.put("statetimems", getCompletionTimestamp());
209     return map;
210   }
211 
212   @Override
213   public String toJSON() throws IOException {
214     return MAPPER.writeValueAsString(toMap());
215   }
216 
217   @Override
218   public String toString() {
219     StringBuilder sb = new StringBuilder(512);
220     sb.append(getDescription());
221     sb.append(": status=");
222     sb.append(getStatus());
223     sb.append(", state=");
224     sb.append(getState());
225     sb.append(", startTime=");
226     sb.append(getStartTime());
227     sb.append(", completionTime=");
228     sb.append(getCompletionTimestamp());
229     return sb.toString();
230   }
231 
232   /**
233    * Returns the status journal. This implementation of status journal is not thread-safe. Currently
234    * we use this to track various stages of flushes and compactions where we can use this/pretty
235    * print for post task analysis, by which time we are already done changing states (writing to
236    * journal)
237    */
238   @Override
239   public List<StatusJournalEntry> getStatusJournal() {
240     if (journal == null) {
241       return Collections.emptyList();
242     } else {
243       return Collections.unmodifiableList(journal);
244     }
245   }
246 
247   /**
248    * Enables journaling of this monitored task, the first invocation will lazily initialize the
249    * journal. The journal implementation itself and this method are not thread safe
250    */
251   @Override
252   public void enableStatusJournal(boolean includeCurrentStatus) {
253     if (journalEnabled && journal != null) {
254       return;
255     }
256     journalEnabled = true;
257     if (journal == null) {
258       journal = new ArrayList<StatusJournalEntry>();
259     }
260     if (includeCurrentStatus) {
261       journal.add(new StatusJournalEntryImpl(status, statusTime));
262     }
263   }
264 
265   @Override
266   public void disableStatusJournal() {
267     journalEnabled = false;
268   }
269 
270   @Override
271   public String prettyPrintJournal() {
272     return StringUtils.join("\n\t", getStatusJournal());
273   }
274 
275 }