import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Override
public void initialize(final TSDB tsdb) {
  this.tsdb = tsdb;
  config = new KafkaRpcPluginConfig(tsdb.getConfig());
  consumer_groups = createConsumerGroups();
  LOG.info("Launching " + consumer_groups.size() + " Kafka consumer groups...");
  for (final KafkaRpcPluginGroup group : consumer_groups) {
    group.start();
  }
  LOG.info("Launched " + consumer_groups.size() + " Kafka consumer groups");
  tsdb.getTimer().newTimeout(this, 100, TimeUnit.MILLISECONDS);
  // Sync just in case the HTTP plugin loads or tries to fetch stats before 
  // we finish initializing.
  synchronized (tsdb) {
    KAFKA_RPC_REFERENCE = this;
  }
  LOG.info("Initialized KafkaRpcPlugin.");
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Override
public synchronized void start() {
  logger.info(String.format("Starting: %s:%s series:%s uids:%s batchSize:%d",
          zkquorum, zkpath, seriesTable, uidsTable, batchSize));
  hbaseClient = new HBaseClient(zkquorum, zkpath);
  try {
      Config config = new Config(false);
      config.overrideConfig("tsd.storage.hbase.data_table", "tsdb");
      config.overrideConfig("tsd.storage.hbase.uid_table", "tsdb-uid");
      config.overrideConfig("tsd.core.auto_create_metrics", "true");
      config.overrideConfig("tsd.storage.enable_compaction", "false");

      tsdb = new TSDB(hbaseClient, config);
  } catch (IOException e) {
      logger.error("tsdb initialization fail: ", e);
  }
  channelCounter.start();
  sinkCounter.start();
  super.start();
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
	 * {@inheritDoc}
	 * @see net.opentsdb.tsd.RTPublisher#initialize(net.opentsdb.core.TSDB)
	 */
	@Override
	public void initialize(final TSDB tsdb) {
		log.info(">>>>> Initializing KafkaRTPublisher...");
		final Map<String, String> config = tsdb.getConfig().getMap();
		final Properties p = new Properties();		
		p.putAll(config);
		final Properties rtpConfig = Props.extract(CONFIG_PREFIX, p, true, true);
		for(final String key: rtpConfig.stringPropertyNames()) {
			rtpConfig.setProperty("kafka.producer." + key , rtpConfig.getProperty(key));
//			rtpConfig.remove(key);
		}
		log.info("EXTRACTED Config:" + rtpConfig);
		kafkaSender = KafkaProducerService.getInstance(rtpConfig);
		messageQueue = MessageQueue.getInstance(getClass().getSimpleName(), this, rtpConfig);
		topicName = rtpConfig.getProperty("kafka.producer.topic.name");
				
		log.info("<<<<< KafkaRTPublisher initialized.");
	}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
	 * Creates a new test TSDB
	 * @param configName The config name to configure with
	 * @return the created test TSDB
	 */
	public static TSDB newTSDB(String configName)  {		
		try {
			tsdb = new TSDB(getConfig(configName));
			tsdb.getConfig().overrideConfig("helios.config.name", configName);
			Config config = tsdb.getConfig();
			StringBuilder b = new StringBuilder("\n\t=============================================\n\tTSDB Config\n\t=============================================");
			for(Map.Entry<String, String> entry: config.getMap().entrySet()) {
				b.append("\n\t").append(entry.getKey()).append("\t:[").append(entry.getValue()).append("]");
			}
			b.append("\n\t=============================================\n");
//			log(b.toString());
			tsdb.initializePlugins(true);
			final UniqueIdRegistry reg = UniqueIdRegistry.getInstance(tsdb);
			tagKunik = reg.getTagKUniqueId();
			tagVunik = reg.getTagVUniqueId();
			tagMunik = reg.getMetricsUniqueId();
			return tsdb;
		} catch (Exception e) {
			throw new RuntimeException("Failed to get test TSDB [" + configName + "]", e);
		}		
	}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Before
public void before() throws Exception {
  tsdb = PowerMockito.mock(TSDB.class);
  config = new Config(false);
  connection_manager = mock(PoolingNHttpClientConnectionManager.class);
  client_builder = mock(HttpAsyncClientBuilder.class);
  client = mock(CloseableHttpAsyncClient.class);
  ts_meta_schema = mock(TSMetaSchema.class);
  uid_meta_schema = mock(UIDMetaSchema.class);
  annotation_schema = mock(AnnotationSchema.class);
  
  config.overrideConfig("tsd.search.elasticsearch.host", "localhost:9200");
  
  when(tsdb.getConfig()).thenReturn(config);
  
  PowerMockito.mockStatic(HttpAsyncClients.class);
  when(HttpAsyncClients.custom()).thenReturn(client_builder);
  
  PowerMockito.whenNew(PoolingNHttpClientConnectionManager.class)
    .withAnyArguments().thenReturn(connection_manager);
  when(client_builder.build()).thenReturn(client);
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Override
public void initialize(TSDB tsdb) {
  this.tsdb = tsdb;
  config = new KafkaRpcPluginConfig(tsdb.getConfig());

  setKafkaConfig();
  producer = new Producer<String, byte[]>(producer_config);
  LOG.info("Initialized kafka requeue publisher.");
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Override
public void execute(final TSDB tsdb, final HttpRpcPluginQuery query) throws IOException {
  // only accept GET/POST for now
  if (query.request().getMethod() != HttpMethod.GET && 
      query.request().getMethod() != HttpMethod.POST) {
    throw new BadRequestException(HttpResponseStatus.METHOD_NOT_ALLOWED, 
        "Method not allowed", "The HTTP method [" + query.method().getName() +
        "] is not permitted for this endpoint");
  }
  
  final String[] uri = query.explodePath();
  final String endpoint = uri.length > 1 ? uri[2].toLowerCase() : "";
  
  if ("version".equals(endpoint)) {
    handleVersion(query);
  } else if ("rate".equals(endpoint)) {
    handleRate(query);
  } else if ("namespace".equals(endpoint)) {
    handlePerNamespaceStats(query);
  } else if ("perthread".equals(endpoint)) {
    handlePerThreadStats(query);
  } else {
    throw new BadRequestException(HttpResponseStatus.NOT_IMPLEMENTED, 
        "Hello. You have reached an API that has been disconnected. "
        + "Please call again.");
  }
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Before
public void before() throws Exception {
  tsdb = PowerMockito.mock(TSDB.class);
  consumer = mock(KafkaRpcPluginThread.class);
  seh = mock(StorageExceptionHandler.class);
  
  when(consumer.getTSDB()).thenReturn(tsdb);
  when(tsdb.getStorageExceptionHandler()).thenReturn(seh);
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Override
public void setReady(TSDB tsdb) {
    LOGGER.debug("OpenTSDB is Ready");
    try {
        register();
    } catch (Exception e) {
        LOGGER.error("Could not register this instance with Consul", e);
    }
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Override
  public void setReady(TSDB tsdb) {
    log.info("OpenTSDB is Ready");
//    Config config = tsdb.getConfig();
//    Integer port = config.getInt("tsd.network.port");
//    log.info("OpenTSDB is listening on " + Integer.toString(port));
    //return;
  }
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
@Override
public void initialize(TSDB tsdb) {
    LOG.debug("Initialized Authentication Plugin");
    this.adminAccessKey = tsdb.getConfig().getString("tsd.core.authentication.admin_access_key");
    this.adminSecretKey = tsdb.getConfig().getString("tsd.core.authentication.admin_access_secret");
    storeCredentials(this.adminAccessKey, this.adminSecretKey);
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
public void initialize(final TSDB tsdb) {
  LOG.info("init RollupPublisher");
  this.tsdb = tsdb;
  this.dataPointsMap = new HashMap<String, DataPoints>();
  Config config = tsdb.getConfig();
  if (config.hasProperty(rollupKey)) {
    this.minutes = tsdb.getConfig().getInt(rollupKey);
  }
  LOG.info("Using window of:" + this.minutes + " minutes");
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
public Deferred<Object> writePoint(TSDB tsdb) {
  Deferred<Object> d;
  if (Tags.looksLikeInteger(value)) {
    d = tsdb.addPoint(metric, timestamp, Tags.parseLong(value), tags);
  } else {  // floating point value
    d = tsdb.addPoint(metric, timestamp, Float.parseFloat(value), tags);
  }
  return d;
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
 * Starts the periodic generation of annotation indexing events using randomly generated values for UIDs.
 * @param tsdb The TSDB to push the annotations to
 * @param quantity The total number of annotations to push
 * @param customs The number of custom map entries per annotation
 * @param period The frequency of publication in ms. Frequencies of less than 1 ms. will push out the entire quantity at once.
 * @return a map of the generated annotations.
 */
public Map<String, Annotation> startAnnotationStream(final TSDB tsdb, int quantity, int customs, final long period) {
	final Map<String, Annotation> annotations = new LinkedHashMap<String, Annotation>(quantity);
	for(int i = 0; i < quantity; i++) {
		Annotation a = new Annotation();
		if(customs>0) {
			HashMap<String, String> custs = new LinkedHashMap<String, String>(customs);
			for(int c = 0; c < customs; c++) {
				String[] frags = getRandomFragments();
				custs.put(frags[0], frags[1]);
			}
			a.setCustom(custs);
		}
		a.setDescription(getRandomFragment());
		long start = nextPosLong();
		a.setStartTime(start);
		a.setEndTime(start + nextPosInt(10000));
		a.setTSUID(getRandomFragment());
		annotations.put(a.getTSUID() + "/" + a.getStartTime(), a);
	}
	Runnable r = new Runnable() {
		@Override
		public void run() {
			try {
				for(Annotation an: annotations.values()) {
					tsdb.indexAnnotation(an);
					if(period>0) {
						Thread.currentThread().join(period);
					}
				}
			} catch (Exception ex) {
				ex.printStackTrace(System.err);
				return;
			}
		}
	};
	startStream(r, "AnnotationStream");
	return annotations;
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * @see net.opentsdb.tsd.HttpRpcPlugin#initialize(net.opentsdb.core.TSDB)
 */
@Override
public void initialize(final TSDB tsdb) {
	log.info(">>>>> Initializing MetricsAPIHttpPlugin....");
	this.tsdb = tsdb;
	findContentDir();
	metricsMetaAPI = HubManager.getInstance().getMetricMetaService();
	jsonMetricsService = new JSONMetricsAPIService(metricsMetaAPI);
	log.info("<<<<< MetricsAPIHttpPlugin Initialized.");
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
	 * {@inheritDoc}
	 * @see net.opentsdb.tsd.HttpRpcPlugin#execute(net.opentsdb.core.TSDB, net.opentsdb.tsd.HttpRpcPluginQuery)
	 */
	@Override
	public void execute(final TSDB tsdb, final HttpRpcPluginQuery query) throws IOException {
		log.info("HTTP Query: [{}]", query.getQueryBaseRoute());
		final String baseURI = query.request().getUri();
		// ==================================================================
		// If baseURI is MAPI_CONTENT, then serve static content
		// Otherwise delegate to metric api service
		// ==================================================================
		
		query.notFound();
//		if(CONTENT_BASE.equals(baseURI)) {
//			sendFile(CONTENT_BASE + "/index.html", 0);
//			return;
//		}
//	    final String uri = baseURI.replace("metricapi-ui/", "");
//	    if ("/favicon.ico".equals(uri)) {
//	      sendFile(staticDir 
//	          + "/favicon.ico", 31536000 /*=1yr*/);
//	      return;
//	    }
//	    if (uri.length() < 3) {  // Must be at least 3 because of the "/s/".
//	      throw new RuntimeException("URI too short <code>" + uri + "</code>");
//	    }
//	    // Cheap security check to avoid directory traversal attacks.
//	    // TODO(tsuna): This is certainly not sufficient.
//	    if (uri.indexOf("..", 3) > 0) {
//	      throw new RuntimeException("Malformed URI <code>" + uri + "</code>");
//	    }
//	    final int questionmark = uri.indexOf('?', 3);
//	    final int pathend = questionmark > 0 ? questionmark : uri.length();
//	    sendFile(contentDir + "/" 
//	                 + uri.substring(1, pathend),  // Drop the "/s"
//	                   uri.contains("nocache") ? 0 : 31536000 /*=1yr*/);
		

	}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * @see net.opentsdb.tsd.RpcPlugin#initialize(net.opentsdb.core.TSDB)
 */
@Override
public void initialize(final TSDB tsdb) {
	log.info(">>>>> Initializing TSDBKafkaEndpointPublisher...");
	final Properties properties = new Properties();
	properties.putAll(tsdb.getConfig().getMap());
	final String zkConnect = properties.getProperty("tsd.storage.hbase.zk_quorum", "localhost:2181");
	log.info("ZK Connect: [{}]", zkConnect);
	System.setProperty("streamhub.discovery.zookeeper.connect", zkConnect);
	final String[] rpcPlugins = ConfigurationHelper.getArraySystemThenEnvProperty("tsd.rpc.plugins", new String[]{}, properties);
	Arrays.sort(rpcPlugins);
	if(Arrays.binarySearch(rpcPlugins, "com.heliosapm.opentsdb.jmx.JMXRPC") >= 0) {
		final String host = AgentName.getInstance().getHostName();
		String jmxmpUri = String.format(DEFAULT_JMX_URLS, host);
		Set<String> endpoints = new HashSet<String>();
		endpoints.add("jvm");
		endpoints.add("tsd");
		if(Arrays.binarySearch(rpcPlugins, "com.heliosapm.streams.opentsdb.KafkaRPC") >= 0) {
			endpoints.add("kafka-consumer");
		}
		if(Arrays.binarySearch(rpcPlugins, "com.heliosapm.streams.opentsdb.KafkaRTPublisher") >= 0) {
			endpoints.add("kafka-producer");
		}			
		EndpointPublisher.getInstance().register(jmxmpUri, endpoints.toArray(new String[endpoints.size()]));
	}
	log.info(">>>>> TSDBEndpointPublisher initialized.");
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
 * Creates a new InternalStatsCollector
 * @param tsdb The tsdb to write the metrics to
 * @param prefix
 */
public InternalStatsCollector(final TSDB tsdb, final String prefix) {
	super(prefix);		
	this.tsdb = tsdb;
	final Map<String, String> tmp = new LinkedHashMap<String, String>();
	final AgentName am = AgentName.getInstance();
	tmp.put("host", am.getHostName());
	tmp.put("app", am.getAppName());
	atags = Collections.unmodifiableMap(tmp);
	
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * @see net.opentsdb.tsd.HttpRpcPlugin#initialize(net.opentsdb.core.TSDB)
 */
@Override
public void initialize(final TSDB tsdb) {
	log.info(">>>>> Initializing WebSocketRPC service....");
	this.tsdb = tsdb;
	webSockHandler = new WebSocketServiceHandler(tsdb);
	final Properties properties = new Properties();
	final Config cfg = tsdb.getConfig();
	properties.putAll(cfg.getMap());		
	path = metricManager.getAndSetConfig(CONFIG_RPC_PATH, DEFAULT_RPC_PATH, properties, cfg);
	JSONRequestRouter.getInstance().registerJSONService(new SystemJSONServices());
	//JSONRequestRouter.getInstance().registerJSONService(new TSDBJSONService(tsdb));

	log.info("<<<<< WebSocketRPC service Initialized.");
}
 

import net.opentsdb.core.TSDB; //导入依赖的package包/类
/**
 * Creates the TSDB instance and plugin jar
 */
@BeforeClass
public static void init() {		
	System.setProperty("tsdb.id.host", "helioleopard");
	createPluginJar(KafkaRPC.class);
	Retransformer.getInstance().transform(TSDB.class, TSDBTestTemplate.class);		
	Retransformer.getInstance().transform(RpcManager.class, FakeRpcManager.class);		
	tsdb = newTSDB("coretest");
	RpcManager.instance(tsdb);
	startProducer();
}