This project has retired. For details please refer to its
Attic page.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.chukwa.datacollection.adaptor.sigar;
20
21 import java.util.HashMap;
22 import java.util.TimerTask;
23
24 import org.apache.commons.lang.exception.ExceptionUtils;
25 import org.apache.hadoop.chukwa.ChunkImpl;
26 import org.apache.hadoop.chukwa.datacollection.ChunkReceiver;
27 import org.apache.hadoop.chukwa.util.ExceptionUtil;
28 import org.apache.log4j.Logger;
29 import org.hyperic.sigar.CpuInfo;
30 import org.hyperic.sigar.CpuPerc;
31 import org.hyperic.sigar.FileSystem;
32 import org.hyperic.sigar.FileSystemUsage;
33 import org.hyperic.sigar.Mem;
34 import org.hyperic.sigar.SigarException;
35 import org.hyperic.sigar.Swap;
36 import org.hyperic.sigar.NetInterfaceStat;
37 import org.hyperic.sigar.Sigar;
38 import org.hyperic.sigar.Uptime;
39 import org.json.simple.JSONArray;
40 import org.json.simple.JSONObject;
41
42
43
44
45 public class SigarRunner extends TimerTask {
46
47 private static Sigar sigar = new Sigar();
48 private static Logger log = Logger.getLogger(SigarRunner.class);
49 private ChunkReceiver receiver = null;
50 private long sendOffset = 0;
51 private SystemMetrics systemMetrics;
52 private HashMap<String, JSONObject> previousNetworkStats = new HashMap<String, JSONObject>();
53 private HashMap<String, JSONObject> previousDiskStats = new HashMap<String, JSONObject>();
54
55 public SigarRunner(ChunkReceiver dest, SystemMetrics systemMetrics) {
56 receiver = dest;
57 this.systemMetrics = systemMetrics;
58 }
59
60 @SuppressWarnings("unchecked")
61 @Override
62 public void run() {
63 boolean skip = false;
64 CpuInfo[] cpuinfo = null;
65 CpuPerc[] cpuPerc = null;
66 Mem mem = null;
67 Swap swap = null;
68 FileSystem[] fs = null;
69 String[] netIf = null;
70 Uptime uptime = null;
71 double[] loadavg = null;
72 JSONObject json = new JSONObject();
73 try {
74
75 JSONArray load = new JSONArray();
76 try {
77 cpuinfo = sigar.getCpuInfoList();
78 cpuPerc = sigar.getCpuPercList();
79 JSONArray cpuList = new JSONArray();
80 for (int i = 0; i < cpuinfo.length; i++) {
81 JSONObject cpuMap = new JSONObject();
82 cpuMap.putAll(cpuinfo[i].toMap());
83 cpuMap.put("combined", cpuPerc[i].getCombined());
84 cpuMap.put("user", cpuPerc[i].getUser());
85 cpuMap.put("sys", cpuPerc[i].getSys());
86 cpuMap.put("idle", cpuPerc[i].getIdle());
87 cpuMap.put("wait", cpuPerc[i].getWait());
88 cpuMap.put("nice", cpuPerc[i].getNice());
89 cpuMap.put("irq", cpuPerc[i].getIrq());
90 cpuList.add(cpuMap);
91 }
92 sigar.getCpuPerc();
93 json.put("cpu", cpuList);
94
95
96 uptime = sigar.getUptime();
97 json.put("uptime", uptime.getUptime());
98
99
100 loadavg = sigar.getLoadAverage();
101 load.add(loadavg[0]);
102 load.add(loadavg[1]);
103 load.add(loadavg[2]);
104 } catch(SigarException se) {
105 log.error("SigarException caused during collection of CPU utilization");
106 log.error(ExceptionUtils.getStackTrace(se));
107 } finally {
108 json.put("loadavg", load);
109 }
110
111
112
113 JSONObject memMap = new JSONObject();
114 JSONObject swapMap = new JSONObject();
115 try {
116 mem = sigar.getMem();
117 memMap.putAll(mem.toMap());
118
119
120 swap = sigar.getSwap();
121 swapMap.putAll(swap.toMap());
122 } catch(SigarException se){
123 log.error("SigarException caused during collection of Memory utilization");
124 log.error(ExceptionUtils.getStackTrace(se));
125 } finally {
126 json.put("memory", memMap);
127 json.put("swap", swapMap);
128 }
129
130
131 JSONArray netInterfaces = new JSONArray();
132 try {
133 netIf = sigar.getNetInterfaceList();
134 for (int i = 0; i < netIf.length; i++) {
135 NetInterfaceStat net = new NetInterfaceStat();
136 try {
137 net = sigar.getNetInterfaceStat(netIf[i]);
138 } catch(SigarException e){
139
140 log.warn("SigarException trying to stat network device "+netIf[i]);
141 continue;
142 }
143 JSONObject netMap = new JSONObject();
144 netMap.putAll(net.toMap());
145 if(previousNetworkStats.containsKey(netIf[i])) {
146 JSONObject deltaMap = previousNetworkStats.get(netIf[i]);
147 deltaMap.put("RxBytes", Long.parseLong(netMap.get("RxBytes").toString()) - Long.parseLong(deltaMap.get("RxBytes").toString()));
148 deltaMap.put("RxDropped", Long.parseLong(netMap.get("RxDropped").toString()) - Long.parseLong(deltaMap.get("RxDropped").toString()));
149 deltaMap.put("RxErrors", Long.parseLong(netMap.get("RxErrors").toString()) - Long.parseLong(deltaMap.get("RxErrors").toString()));
150 deltaMap.put("RxPackets", Long.parseLong(netMap.get("RxPackets").toString()) - Long.parseLong(deltaMap.get("RxPackets").toString()));
151 deltaMap.put("TxBytes", Long.parseLong(netMap.get("TxBytes").toString()) - Long.parseLong(deltaMap.get("TxBytes").toString()));
152 deltaMap.put("TxCollisions", Long.parseLong(netMap.get("TxCollisions").toString()) - Long.parseLong(deltaMap.get("TxCollisions").toString()));
153 deltaMap.put("TxErrors", Long.parseLong(netMap.get("TxErrors").toString()) - Long.parseLong(deltaMap.get("TxErrors").toString()));
154 deltaMap.put("TxPackets", Long.parseLong(netMap.get("TxPackets").toString()) - Long.parseLong(deltaMap.get("TxPackets").toString()));
155 netInterfaces.add(deltaMap);
156 skip = false;
157 } else {
158 netInterfaces.add(netMap);
159 skip = true;
160 }
161 previousNetworkStats.put(netIf[i], netMap);
162 }
163 } catch(SigarException se){
164 log.error("SigarException caused during collection of Network utilization");
165 log.error(ExceptionUtils.getStackTrace(se));
166 } finally {
167 json.put("network", netInterfaces);
168 }
169
170
171 JSONArray fsList = new JSONArray();
172 try {
173 fs = sigar.getFileSystemList();
174 for (int i = 0; i < fs.length; i++) {
175 FileSystemUsage usage = sigar.getFileSystemUsage(fs[i].getDirName());
176 JSONObject fsMap = new JSONObject();
177 fsMap.putAll(fs[i].toMap());
178 fsMap.put("ReadBytes", usage.getDiskReadBytes());
179 fsMap.put("Reads", usage.getDiskReads());
180 fsMap.put("WriteBytes", usage.getDiskWriteBytes());
181 fsMap.put("Writes", usage.getDiskWrites());
182 if(previousDiskStats.containsKey(fs[i].getDevName())) {
183 JSONObject deltaMap = previousDiskStats.get(fs[i].getDevName());
184 deltaMap.put("ReadBytes", usage.getDiskReadBytes() - (Long) deltaMap.get("ReadBytes"));
185 deltaMap.put("Reads", usage.getDiskReads() - (Long) deltaMap.get("Reads"));
186 deltaMap.put("WriteBytes", usage.getDiskWriteBytes() - (Long) deltaMap.get("WriteBytes"));
187 deltaMap.put("Writes", usage.getDiskWrites() - (Long) deltaMap.get("Writes"));
188 deltaMap.put("Total", usage.getTotal());
189 deltaMap.put("Used", usage.getUsed());
190 deltaMap.putAll(fs[i].toMap());
191 fsList.add(deltaMap);
192 skip = false;
193 } else {
194 fsList.add(fsMap);
195 skip = true;
196 }
197 previousDiskStats.put(fs[i].getDevName(), fsMap);
198 }
199 } catch(SigarException se){
200 log.error("SigarException caused during collection of FileSystem utilization");
201 log.error(ExceptionUtils.getStackTrace(se));
202 } finally {
203 json.put("disk", fsList);
204 }
205 json.put("timestamp", System.currentTimeMillis());
206 byte[] data = json.toString().getBytes();
207 sendOffset += data.length;
208 ChunkImpl c = new ChunkImpl("SystemMetrics", "Sigar", sendOffset, data, systemMetrics);
209 if(!skip) {
210 receiver.add(c);
211 }
212 } catch (Exception se) {
213 log.error(ExceptionUtil.getStackTrace(se));
214 }
215 }
216
217 }