datasource.java

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

/* ============================================================
 * JRobin : Pure java implementation of RRDTool's functionality
 * ============================================================
 *
 * Project Info:  http://www.jrobin.org
 * Project Lead:  Sasa Markovic ([e-mail]);
 *
 * (C) Copyright 2003-2005, by Sasa Markovic.
 *
 * Developers:    Sasa Markovic ([e-mail])
 *
 *
 * This library is free software; you can redistribute it and/or modify it under the terms
 * of the GNU Lesser General Public License as published by the Free Software Foundation;
 * either version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
 * See the GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License along with this
 * library; if not, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307, USA.
 */

package org.jrobin.core;

import java.io.IOException;

/**
 * Class to represent single datasource within RRD. Each datasource object holds the
 * following information: datasource definition (once set, never changed) and
 * datasource state variables (changed whenever RRD gets updated).<p>
 * <p/>
 * Normally, you don't need to manipluate Datasource objects directly, it's up to
 * JRobin framework to do it for you.
 *
 * @author <a href="mailto:[e-mail]">Sasa Markovic</a>
 */

public class Datasource implements RrdUpdater, DsTypes {
	private static final double MAX_32_BIT = Math.pow(2, 32);
	private static final double MAX_64_BIT = Math.pow(2, 64);

	private RrdDb parentDb;
	// definition
	private RrdString dsName, dsType;
	private RrdLong heartbeat;
	private RrdDouble minValue, maxValue;

	// state variables
	private RrdDouble lastValue;
	private RrdLong nanSeconds;
	private RrdDouble accumValue;

	Datasource(RrdDb parentDb, DsDef dsDef) throws IOException {
		boolean shouldInitialize = dsDef != null;
		this.parentDb = parentDb;
		dsName = new RrdString(this);
		dsType = new RrdString(this);
		heartbeat = new RrdLong(this);
		minValue = new RrdDouble(this);
		maxValue = new RrdDouble(this);
		lastValue = new RrdDouble(this);
		accumValue = new RrdDouble(this);
		nanSeconds = new RrdLong(this);
		if (shouldInitialize) {
			dsName.set(dsDef.getDsName());
			dsType.set(dsDef.getDsType());
			heartbeat.set(dsDef.getHeartbeat());
			minValue.set(dsDef.getMinValue());
			maxValue.set(dsDef.getMaxValue());
			lastValue.set(Double.NaN);
			accumValue.set(0.0);
			Header header = parentDb.getHeader();
			nanSeconds.set(header.getLastUpdateTime() % header.getStep());
		}
	}

	Datasource(RrdDb parentDb, DataImporter reader, int dsIndex) throws IOException, RrdException {
		this(parentDb, null);
		dsName.set(reader.getDsName(dsIndex));
		dsType.set(reader.getDsType(dsIndex));
		heartbeat.set(reader.getHeartbeat(dsIndex));
		minValue.set(reader.getMinValue(dsIndex));
		maxValue.set(reader.getMaxValue(dsIndex));
		lastValue.set(reader.getLastValue(dsIndex));
		accumValue.set(reader.getAccumValue(dsIndex));
		nanSeconds.set(reader.getNanSeconds(dsIndex));
	}

	String dump() throws IOException {
		return "== DATASOURCE ==\n" +
				"DS:" + dsName.get() + ":" + dsType.get() + ":" +
				heartbeat.get() + ":" + minValue.get() + ":" +
				maxValue.get() + "\nlastValue:" + lastValue.get() +
				" nanSeconds:" + nanSeconds.get() +
				" accumValue:" + accumValue.get() + "\n";
	}

	/**
	 * Returns datasource name.
	 *
	 * @return Datasource name
	 * @throws IOException Thrown in case of I/O error
	 */
	public String getDsName() throws IOException {
		return dsName.get();
	}

	/**
	 * Returns datasource type (GAUGE, COUNTER, DERIVE, ABSOLUTE).
	 *
	 * @return Datasource type.
	 * @throws IOException Thrown in case of I/O error
	 */
	public String getDsType() throws IOException {
		return dsType.get();
	}

	/**
	 * Returns datasource heartbeat
	 *
	 * @return Datasource heartbeat
	 * @throws IOException Thrown in case of I/O error
	 */

	public long getHeartbeat() throws IOException {
		return heartbeat.get();
	}

	/**
	 * Returns mimimal allowed value for this datasource.
	 *
	 * @return Minimal value allowed.
	 * @throws IOException Thrown in case of I/O error
	 */
	public double getMinValue() throws IOException {
		return minValue.get();
	}

	/**
	 * Returns maximal allowed value for this datasource.
	 *
	 * @return Maximal value allowed.
	 * @throws IOException Thrown in case of I/O error
	 */
	public double getMaxValue() throws IOException {
		return maxValue.get();
	}

	/**
	 * Returns last known value of the datasource.
	 *
	 * @return Last datasource value.
	 * @throws IOException Thrown in case of I/O error
	 */
	public double getLastValue() throws IOException {
		return lastValue.get();
	}

	/**
	 * Returns value this datasource accumulated so far.
	 *
	 * @return Accumulated datasource value.
	 * @throws IOException Thrown in case of I/O error
	 */
	public double getAccumValue() throws IOException {
		return accumValue.get();
	}

	/**
	 * Returns the number of accumulated NaN seconds.
	 *
	 * @return Accumulated NaN seconds.
	 * @throws IOException Thrown in case of I/O error
	 */
	public long getNanSeconds() throws IOException {
		return nanSeconds.get();
	}

	void process(long newTime, double newValue) throws IOException, RrdException {
		Header header = parentDb.getHeader();
		long step = header.getStep();
		long oldTime = header.getLastUpdateTime();
		long startTime = Util.normalize(oldTime, step);
		long endTime = startTime + step;
		double oldValue = lastValue.get();
		double updateValue = calculateUpdateValue(oldTime, oldValue, newTime, newValue);
		if (newTime < endTime) {
			accumulate(oldTime, newTime, updateValue);
		}
		else {
			// should store something
			long boundaryTime = Util.normalize(newTime, step);
			accumulate(oldTime, boundaryTime, updateValue);
			double value = calculateTotal(startTime, boundaryTime);
			// how many updates?
			long numSteps = (boundaryTime - endTime) / step + 1L;
			// ACTION!
			parentDb.archive(this, value, numSteps);
			// cleanup
			nanSeconds.set(0);
			accumValue.set(0.0);
			accumulate(boundaryTime, newTime, updateValue);
		}
	}

	private double calculateUpdateValue(long oldTime, double oldValue,
										long newTime, double newValue) throws IOException {
		double updateValue = Double.NaN;
		if (newTime - oldTime <= heartbeat.get()) {
			String type = dsType.get();
			if (type.equals(DT_GAUGE)) {
				updateValue = newValue;
			}
			else if (type.equals(DT_ABSOLUTE)) {
				if (!Double.isNaN(newValue)) {
					updateValue = newValue / (newTime - oldTime);
				}
			}
			else if (type.equals(DT_DERIVE)) {
				if (!Double.isNaN(newValue) && !Double.isNaN(oldValue)) {
					updateValue = (newValue - oldValue) / (newTime - oldTime);
				}
			}
			else if (type.equals(DT_COUNTER)) {
				if (!Double.isNaN(newValue) && !Double.isNaN(oldValue)) {
					double diff = newValue - oldValue;
					if (diff < 0) {
						diff += MAX_32_BIT;
					}
					if (diff < 0) {
						diff += MAX_64_BIT - MAX_32_BIT;
					}
					if (diff >= 0) {
						updateValue = diff / (newTime - oldTime);
					}
				}
			}
			if (!Double.isNaN(updateValue)) {
				double minVal = minValue.get();
				double maxVal = maxValue.get();
				if (!Double.isNaN(minVal) && updateValue < minVal) {