dataprocessor.java

jrobin,使用纯java实现的RRD数据库,使用RRD数据库来统计数据.

	/**
	 * Calculates timestamps which correspond to individual pixels on the graph.
	 *
	 * @param pixelCount Graph width
	 * @return Array of timestamps
	 */
	public long[] getTimestampsPerPixel(int pixelCount) {
		setPixelCount(pixelCount);
		return getTimestampsPerPixel();
	}

	/**
	 * Calculates timestamps which correspond to individual pixels on the graph
	 * based on the graph width set with a {@link #setPixelCount(int)} method call.
	 *
	 * @return Array of timestamps
	 */
	public long[] getTimestampsPerPixel() {
		long[] times = new long[pixelCount];
		long span = tEnd - tStart;
		for (int i = 0; i < pixelCount; i++) {
			times[i] = Math.round(tStart + (double) (span * i) / (double) (pixelCount - 1));
		}
		return times;
	}

	/**
	 * Dumps timestamps and values of all datasources in a tabelar form. Very useful for debugging.
	 *
	 * @return Dumped object content.
	 * @throws RrdException Thrown if nothing is calculated so far (the method {@link #processData()}
	 *                      was not called).
	 */
	public String dump() throws RrdException {
		String[] names = getSourceNames();
		double[][] values = getValues();
		StringBuffer buffer = new StringBuffer();
		buffer.append(format("timestamp", 12));
		for (String name : names) {
			buffer.append(format(name, 20));
		}
		buffer.append("\n");
		for (int i = 0; i < timestamps.length; i++) {
			buffer.append(format("" + timestamps[i], 12));
			for (int j = 0; j < names.length; j++) {
				buffer.append(format(Util.formatDouble(values[j][i]), 20));
			}
			buffer.append("\n");
		}
		return buffer.toString();
	}

	/**
	 * Returns time when last RRD archive was updated (all RRD files are considered).
	 *
	 * @return Last archive update time for all RRD files in this DataProcessor
	 */
	public long getLastRrdArchiveUpdateTime() {
		return lastRrdArchiveUpdateTime;
	}

	// PRIVATE METHODS

	private void extractDefs() {
		List<Def> defList = new ArrayList<Def>();
		for (Source source : sources.values()) {
			if (source instanceof Def) {
				defList.add((Def) source);
			}
		}
		defSources = defList.toArray(new Def[defList.size()]);
	}

	private void fetchRrdData() throws IOException, RrdException {
		long tEndFixed = (tEnd == 0) ? Util.getTime() : tEnd;
		for (int i = 0; i < defSources.length; i++) {
			if (!defSources[i].isLoaded()) {
				// not fetched yet
				Set<String> dsNames = new HashSet<String>();
				dsNames.add(defSources[i].getDsName());
				// look for all other datasources with the same path and the same consolidation function
				for (int j = i + 1; j < defSources.length; j++) {
					if (defSources[i].isCompatibleWith(defSources[j])) {
						dsNames.add(defSources[j].getDsName());
					}
				}
				// now we have everything
				RrdDb rrd = null;
				try {
					rrd = getRrd(defSources[i]);
					lastRrdArchiveUpdateTime = Math.max(lastRrdArchiveUpdateTime, rrd.getLastArchiveUpdateTime());
					FetchRequest req = rrd.createFetchRequest(defSources[i].getConsolFun(),
							tStart, tEndFixed, fetchRequestResolution);
					req.setFilter(dsNames);
					FetchData data = req.fetchData();
					defSources[i].setFetchData(data);
					for (int j = i + 1; j < defSources.length; j++) {
						if (defSources[i].isCompatibleWith(defSources[j])) {
							defSources[j].setFetchData(data);
						}
					}
				}
				finally {
					if (rrd != null) {
						releaseRrd(rrd, defSources[i]);
					}
				}
			}
		}
	}

	private void fixZeroEndingTimestamp() throws RrdException {
		if (tEnd == 0) {
			if (defSources.length == 0) {
				throw new RrdException("Could not adjust zero ending timestamp, no DEF source provided");
			}
			tEnd = defSources[0].getArchiveEndTime();
			for (int i = 1; i < defSources.length; i++) {
				tEnd = Math.min(tEnd, defSources[i].getArchiveEndTime());
			}
			if (tEnd <= tStart) {
				throw new RrdException("Could not resolve zero ending timestamp.");
			}
		}
	}

	// Tricky and ugly. Should be redesigned some time in the future
	private void chooseOptimalStep() {
		long newStep = Long.MAX_VALUE;
		for (Def defSource : defSources) {
			long fetchStep = defSource.getFetchStep(), tryStep = fetchStep;
			if (step > 0) {
				tryStep = Math.min(newStep, (((step - 1) / fetchStep) + 1) * fetchStep);
			}
			newStep = Math.min(newStep, tryStep);
		}
		if (newStep != Long.MAX_VALUE) {
			// step resolved from a RRD file
			step = newStep;
		}
		else {
			// choose step based on the number of pixels (useful for plottable datasources)
			step = Math.max((tEnd - tStart) / pixelCount, 1);
		}
	}

	private void createTimestamps() {
		long t1 = Util.normalize(tStart, step);
		long t2 = Util.normalize(tEnd, step);
		if (t2 < tEnd) {
			t2 += step;
		}
		int count = (int) (((t2 - t1) / step) + 1);
		timestamps = new long[count];
		for (int i = 0; i < count; i++) {
			timestamps[i] = t1;
			t1 += step;
		}
	}

	private void assignTimestampsToSources() {
		for (Source src : sources.values()) {
			src.setTimestamps(timestamps);
		}
	}

	private void normalizeRrdValues() throws RrdException {
		Normalizer normalizer = new Normalizer(timestamps);
		for (Def def : defSources) {
			long[] rrdTimestamps = def.getRrdTimestamps();
			double[] rrdValues = def.getRrdValues();
			double[] values = normalizer.normalize(rrdTimestamps, rrdValues);
			def.setValues(values);
		}
	}

	private void calculateNonRrdSources() throws RrdException {
		for (Source source : sources.values()) {
			if (source instanceof SDef) {
				calculateSDef((SDef) source);
			}
			else if (source instanceof CDef) {
				calculateCDef((CDef) source);
			}
			else if (source instanceof PDef) {
				calculatePDef((PDef) source);
			}
		}
	}

	private void calculatePDef(PDef pdef) {
		pdef.calculateValues();
	}

	private void calculateCDef(CDef cDef) throws RrdException {
		RpnCalculator calc = new RpnCalculator(cDef.getRpnExpression(), cDef.getName(), this);
		cDef.setValues(calc.calculateValues());
	}

	private void calculateSDef(SDef sDef) throws RrdException {
		String defName = sDef.getDefName();
		String consolFun = sDef.getConsolFun();
		Source source = getSource(defName);
		double value = source.getAggregates(tStart, tEnd).getAggregate(consolFun);
		sDef.setValue(value);
	}

	private RrdDb getRrd(Def def) throws IOException, RrdException {
		String path = def.getPath(), backend = def.getBackend();
		if (poolUsed && backend == null) {
			return RrdDbPool.getInstance().requestRrdDb(path);
		}
		else if (backend != null) {
			return new RrdDb(path, true, RrdBackendFactory.getFactory(backend));
		}
		else {
			return new RrdDb(path, true);
		}
	}

	private void releaseRrd(RrdDb rrd, Def def) throws IOException, RrdException {
		String backend = def.getBackend();
		if (poolUsed && backend == null) {
			RrdDbPool.getInstance().release(rrd);
		}
		else {
			rrd.close();
		}
	}

	private static String format(String s, int length) {
		StringBuffer b = new StringBuffer(s);
		for (int i = 0; i < length - s.length(); i++) {
			b.append(' ');
		}
		return b.toString();
	}

	/**
	 * Cute little demo. Uses demo.rrd file previously created by basic JRobin demo.
	 *
	 * @param args Not used
	 * @throws IOException
	 * @throws RrdException
	 */
	public static void main(String[] args) throws IOException, RrdException {
		// time span
		long t1 = Util.getTimestamp(2003, 4, 1);
		long t2 = Util.getTimestamp(2003, 5, 1);
		System.out.println("t1 = " + t1);
		System.out.println("t2 = " + t2);

		// RRD file to use
		String rrdPath = Util.getJRobinDemoPath("demo.rrd");

		// constructor
		DataProcessor dp = new DataProcessor(t1, t2);

		// uncomment and run again
		//dp.setFetchRequestResolution(86400);

		// uncomment and run again
		//dp.setStep(86500);

		// datasource definitions
		dp.addDatasource("X", rrdPath, "sun", "AVERAGE");
		dp.addDatasource("Y", rrdPath, "shade", "AVERAGE");
		dp.addDatasource("Z", "X,Y,+,2,/");
		dp.addDatasource("DERIVE[Z]", "Z,PREV(Z),-,STEP,/");
		dp.addDatasource("TREND[Z]", "DERIVE[Z],SIGN");
		dp.addDatasource("AVG[Z]", "Z", "AVERAGE");
		dp.addDatasource("DELTA", "Z,AVG[Z],-");

		// action
		long laptime = System.currentTimeMillis();
		//dp.setStep(86400);
		dp.processData();
		System.out.println("Data processed in " + (System.currentTimeMillis() - laptime) + " milliseconds\n---");
		System.out.println(dp.dump());

		// aggregates
		System.out.println("\nAggregates for X");
		Aggregates agg = dp.getAggregates("X");
		System.out.println(agg.dump());
		System.out.println("\nAggregates for Y");
		agg = dp.getAggregates("Y");
		System.out.println(agg.dump());

		// 95-percentile
		System.out.println("\n95-percentile for X: " + Util.formatDouble(dp.get95Percentile("X")));
		System.out.println("95-percentile for Y: " + Util.formatDouble(dp.get95Percentile("Y")));

		// lastArchiveUpdateTime
		System.out.println("\nLast archive update time was: " + dp.getLastRrdArchiveUpdateTime());
	}
}