querynetbox.java

Network Administration Visualized 网络管理可视化源码

import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.Stack;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeMap;

import no.ntnu.nav.ConfigParser.ConfigParser;
import no.ntnu.nav.Database.Database;
import no.ntnu.nav.SimpleSnmp.SimpleSnmp;
import no.ntnu.nav.SimpleSnmp.TimeoutException;
import no.ntnu.nav.event.Event;
import no.ntnu.nav.event.EventQ;
import no.ntnu.nav.event.EventQListener;
import no.ntnu.nav.getDeviceData.Netbox;
import no.ntnu.nav.getDeviceData.dataplugins.DataHandler;
import no.ntnu.nav.getDeviceData.deviceplugins.DeviceHandler;
import no.ntnu.nav.logger.Log;
import no.ntnu.nav.netboxinfo.NetboxInfo;

/**
 * This class schedules the netboxes, assigns them to threads and runs
 * the plugins.
 */

public class QueryNetbox extends Thread
{
	private static ConfigParser navCp;
	private static ConfigParser myCp;
	private static Map dataClassMap, deviceClassMap;

	private static Timer timer;
	private static Timer updateDataTimer;
	private static int updateDataInterval;
	private static CheckRunQTask checkRunQTask;
	private static UpdateDataTask updateDataTask;

	private static Map typeidMap;
	private static Map oidkeyMap;
	private static List initialQ;
	private static SortedMap nbRunQ;
	private static Map netboxidRunQMap;
	private static Stack idleThreads;
	private static Map nbMap;
	private static List nbList;

	private static LinkedList oidQ;
	private static Set oidNetboxSet = Collections.synchronizedSet(new HashSet());
	private static Set oidKeySet = Collections.synchronizedSet(new HashSet());

	private static int maxThreadCnt;
	private static int extraThreadCnt;
	private static Map scheduleImmediatelyMap = Collections.synchronizedMap(new HashMap());
	private static int threadCnt;
	private static Integer idleThreadLock = new Integer(0);
	private static LinkedList threadIdList = new LinkedList();
	private static int curMaxTid = 0;


	private static int netboxCnt;
	private static int netboxHigh;
	private static long nbProcessedCnt;

	private static String qNetbox;

	// Plugins

	// Caches which device handlers can handle a given Netbox
	static Map deviceNetboxCache = Collections.synchronizedMap(new HashMap());

	// Stores the persistent storage for the dataplugins
	static Map persistentStorage = Collections.synchronizedMap(new HashMap());

	// Object data
	String tid;
	NetboxImpl nb;
	Object oidUpdObj;

	// Static init
	public static void init(int numThreads, int updateDataIntervalI, ConfigParser _myCp, ConfigParser _navCp, Map dataCM, Map deviceCM, String qnb) {
		maxThreadCnt = numThreads;
		updateDataInterval = updateDataIntervalI;
		myCp = _myCp;
		navCp = _navCp;
		myCp.setObject("navCp", navCp);
		dataClassMap = dataCM;
		deviceClassMap = deviceCM;
		qNetbox = qnb;

		// Create the netbox map and the run queue
		nbMap = new HashMap();
		initialQ = Collections.synchronizedList(new ArrayList());
		nbRunQ = new TreeMap();
		nbList = new ArrayList();
		netboxidRunQMap = new HashMap();
		oidQ = new LinkedList();

		// Create the EventListener
		EventQ.init(2000);
		EventQ.addEventQListener("getDeviceData", new EventListener());

		timer = new Timer();
		updateDataTimer = new Timer();

		// Fetch from DB
		updateTypes(false);
		updateNetboxes();

		// Schedule fetch updates
		Log.d("INIT", "Starting timer for data updating");
		scheduleUpdateNetboxes(updateDataInterval);

		Log.d("INIT", "Starting timer for netbox query scheduling");
		scheduleCheckRunQ(0);

	}

	private static void scheduleUpdateNetboxes(long l) {
		synchronized (updateDataTimer) {
			if (updateDataTask != null) updateDataTask.cancel();
			updateDataTask = new UpdateDataTask();

			// The delay can actually be negative due to inaccuracy in the Java timer
			l = Math.max(l, 0);
			Log.d("QUERY_NETBOX", "SCHEDULE_UPDATE_NETBOXES", "Schedule update netboxes in " + l + " ms");
			updateDataTimer.schedule(updateDataTask, l, updateDataInterval);
		}
	}

	private static void createUnknownType(Map typeidMapL) {
		// The unknown type is used for netboxes with missing type
		String typeid = Type.UNKNOWN_TYPEID;
		String typename = "unknownType";
		String vendorid = "unknownVendor";
		int csAtVlan = Type.CS_AT_VLAN_UNKNOWN;
		boolean chassis = true;

		Type t = new Type(typeid, typename, vendorid, csAtVlan, chassis);
		typeidMapL.put(typeid, t);

	}

	private static void addDefaultOids(Map keyFreqMap, Map keyMap) {
		String[] defOidKeys = {
			"typeoid",
			"dnscheck",
		};

		for (int i=0; i < defOidKeys.length; i++) {
			String oidkey = defOidKeys[i];
			Snmpoid snmpoid = (Snmpoid)oidkeyMap.get(oidkey);

			keyFreqMap.put(oidkey, new Integer(OidTester.DEFAULT_FREQ));
			keyMap.put(oidkey, snmpoid);
		}
	}

	/*
	// Add the given oidkey
	private static void addOid(String oidkey, Map keyFreqMap, Map keyMap) {		
		try {
			ResultSet rs = Database.query("SELECT snmpoidid, oidkey, snmpoid, getnext, decodehex, match_regex, uptodate FROM snmpoid WHERE oidkey='" + oidkey + "'");
			rs.next();
			keyFreqMap.put(rs.getString("oidkey"), new Integer(OidTester.DEFAULT_FREQ));

			Snmpoid snmpoid = new Snmpoid(rs.getString("snmpoidid"), rs.getString("oidkey"), rs.getString("snmpoid"), rs.getBoolean("getnext"), rs.getBoolean("decodehex"), rs.getString("match_regex"), rs.getBoolean("uptodate"));
			keyMap.put(rs.getString("oidkey"), snmpoid);

		} catch (SQLException e) {
			Log.w("QUERY_NETBOX", "ADD_OID", "Missing oidkey " + oidkey + " from snmpoid, cannot update types!");
			Log.d("QUERY_NETBOX", "ADD_OID", "SQLException: " + e.getMessage());
		}
	}
	*/

	private static void scheduleCheckRunQ(long l)
	{
		synchronized (timer) {
			if (checkRunQTask != null) checkRunQTask.cancel();
			checkRunQTask = new CheckRunQTask();

			// The delay can actually be negative due to inaccuracy in the Java timer
			l = Math.max(l, 0);
			Log.d("QUERY_NETBOX", "SCHEDULE_CHECK_RUN_Q", "Schedule check runq in " + l + " ms");
			try {
				timer.schedule(checkRunQTask, l);
			} catch (IllegalStateException e) {
				timer = new Timer();
				checkRunQTask = new CheckRunQTask();
				timer.schedule(checkRunQTask, l);
			}
		}
	}

	private static void checkRunQ()
	{
		Log.setDefaultSubsystem("QUERY_NETBOX");

		// First we check if the OID database needs updating
		synchronized (oidQ) {
			while (!oidQ.isEmpty()) {
				Object updateO = oidQ.removeFirst();
				Log.d("CHECK_RUN_Q", "oidQ not empty, got: " + updateO);

				// Try to get a free thread
				String tid = requestThread(false);
				if (tid == null) {
					Log.d("CHECK_RUN_Q", "oidQ not empty, but no thread available");
					oidQ.addFirst(updateO);
					return;
				}

				// OK, start a new QueryNetbox
				Log.d("CHECK_RUN_Q", "Starting new OID thread with id: " + tid);
				new QueryNetbox(tid, updateO).start();
			}
		}

		// Try to get a free netbox
		Object o;
		while ((o = removeRunQHead()) instanceof NetboxImpl) {
			NetboxImpl nb = (NetboxImpl)o;

			Log.d("CHECK_RUN_Q", "Got netbox: " + nb);

			// Try to get a free thread
			String tid = requestThread(true);
			if (tid == null) {
				Log.d("CHECK_RUN_Q", "Netbox is available, but no threads are idle");
				// Re-insert into queue
				addToRunQFront(nb);
				return;
			}

			// OK, start a new QueryNetbox
			Log.d("CHECK_RUN_Q", "Starting new Netbox thread with id: " + tid + " to handle " + nb);
			new QueryNetbox(tid, nb).start();

		} 

		// No more free netboxes, schedule re-run when the next is ready
		Long nextRun = (Long)o;
		Log.d("CHECK_RUN_Q", "No available netbox, scheduling next check in " + nextRun + " ms");			
		scheduleCheckRunQ(nextRun.longValue());

	}

	public static synchronized void updateTypes(boolean updateNetboxes) {
		Map typeidM = new HashMap();
		Map oidkeyM = Collections.synchronizedMap(new HashMap());
		int oidnew=0;

		// The unknown type is for netboxes without type
		createUnknownType(typeidM);

		try {
			// First fetch new types from the database
			ResultSet rs = Database.query("SELECT typeid, typename, vendorid, cs_at_vlan, chassis FROM type");
			while (rs.next()) {
				String typeid = rs.getString("typeid");
				String typename = rs.getString("typename");
				int csAtVlan = rs.getString("cs_at_vlan") == null ? Type.CS_AT_VLAN_UNKNOWN : Type.csAtVlan(rs.getBoolean("cs_at_vlan"));
				boolean chassis = rs.getBoolean("chassis");
				
				Type t = new Type(typeid, typename, rs.getString("vendorid"), csAtVlan, chassis);
				typeidM.put(typeid, t);

			}

			// Now fetch all OIDs
			rs = Database.query("SELECT snmpoidid, oidkey, snmpoid, getnext, decodehex, match_regex, defaultfreq, uptodate AS oiduptodate, mib FROM snmpoid");
			while (rs.next()) {
				String snmpoidid = rs.getString("snmpoidid");
				String oidkey = rs.getString("oidkey");
				String oid = rs.getString("snmpoid");
				boolean getnext = rs.getBoolean("getnext");
				boolean decodehex = rs.getBoolean("decodehex");
				String matchRegex = rs.getString("match_regex");
				int defaultfreq = rs.getInt("defaultfreq");
				boolean oiduptodate = rs.getBoolean("oiduptodate");
				String mib = rs.getString("mib");
				
				Snmpoid snmpoid = new Snmpoid(snmpoidid, oidkey, oid, getnext, decodehex, matchRegex, defaultfreq, oiduptodate, mib);
				oidkeyM.put(oidkey, snmpoid);
				if (!oiduptodate && oidKeySet.add(oidkey)) {
					OidTester.clearDupe(snmpoid);
					synchronized (oidQ) {
						oidQ.add(snmpoid);
						oidnew++;
					}
				}
			}

			// Make new types global
			typeidMap = typeidM;
			oidkeyMap = oidkeyM;

			Log.i("UPDATE_TYPES", "Num types: " + typeidMap.size() + ", num OIDs: " + oidkeyMap.size() + ", new oidQ: " + oidnew);

			// Then update all netboxes with the new types
			if (updateNetboxes) updateNetboxesWithNewTypes();

		} catch (SQLException e) {
			Log.e("UPDATE_TYPES", "SQLException: " + e);
			e.printStackTrace(System.err);
		}
	}

	private static synchronized void updateNetboxesWithNewTypes() {
		for (Iterator it = nbMap.values().iterator(); it.hasNext();) {
			NetboxImpl nb = (NetboxImpl)it.next();
			Type t = (Type)typeidMap.get(nb.getTypeT().getTypeid());
			nb.setType(t);
		} 
	}

	public static synchronized void updateNetboxes() {
		int newcnt=0, skipcnt=0, delcnt=0;

		try {
			Map numInStackMap = new HashMap();
			ResultSet rs = Database.query("SELECT netboxid,COUNT(*) AS numInStack FROM module GROUP BY netboxid HAVING COUNT(*) > 1");
			while (rs.next()) numInStackMap.put(rs.getString("netboxid"), rs.getString("numInStack"));

			//String sql = "SELECT ip,ro,deviceid,netboxid,catid,sysname,typeid,typename FROM netbox LEFT JOIN type USING(typeid) WHERE up='y' AND ro IS NOT NULL";
			String sql = "SELECT ip,ro,deviceid,netboxid,catid,sysname,typeid,snmp_version,netbox.uptodate,type.frequency AS typefreq, netboxsnmpoid.frequency AS oidfreq, snmpoidid, oidkey, snmpoid FROM netbox LEFT JOIN type USING(typeid) LEFT JOIN netboxsnmpoid USING(netboxid) LEFT JOIN snmpoid USING(snmpoidid) WHERE up='y' AND ro IS NOT NULL";
			boolean randomize = true;
			if (qNetbox != null) {
				String qn = qNetbox;

				if (qn.startsWith("_") || qn.startsWith("-") || qn.indexOf(",") >= 0) {
					if (qn.startsWith("__")) {
						qn = "";
						sql += " AND netbox.uptodate=false";
					} else if (qn.startsWith("-")) {
						qn = qn.substring(1, qn.length());
						sql += " AND typeid IN (";
					} else if (qn.startsWith("_")) {
						qn = qn.substring(1, qn.length());
						sql += " AND catid IN (";
					} else {
						sql += " AND sysname IN (";
						randomize = false;
					}
					if (qn.length() > 0) {
						String[] ids = qn.split(",");
						for (int i=0; i < ids.length; i++) sql += "'" + ids[i] + "',";
						if (ids.length > 0) sql = sql.substring(0, sql.length()-1);
						sql += ")";
					}
				} else {
					sql += " AND sysname LIKE '"+qn+"'";
				}
			}