可以直接通过源代码包中net.opentsdb.tools.TSDMain类的main方法直接启动，不需要附带任何参数。
public static void main(String[] args) throws IOException {
Logger log = LoggerFactory.getLogger(TSDMain.class);
log.info("Starting.");
log.info(BuildData.revisionString());
log.info(BuildData.buildString());
try {
System.in.close(); // Release a FD we don't need.
} catch (Exception e) {
log.warn("Failed to close stdin", e);
}
final ArgP argp = new ArgP();
CliOptions.addCommon(argp);
argp.addOption("--port", "NUM", "TCP port to listen on.");
...
CliOptions.addAutoMetricFlag(argp);
args = CliOptions.parse(argp, args);
args = null; // free().
// get a config object
Config config = CliOptions.getConfig(argp);
// check for the required parameters
...
final ServerSocketChannelFactory factory;
int connections_limit = ;
try {
connections_limit = config.getInt("tsd.core.connections.limit");
} catch (NumberFormatException nfe) {
usage(argp, "Invalid connections limit", 1);
}
if (config.getBoolean("tsd.network.async_io")) {
int workers = Runtime.getRuntime().availableProcessors() * 2;
if (config.hasProperty("tsd.network.worker_threads")) {
try {
workers = config.getInt("tsd.network.worker_threads");
} catch (NumberFormatException nfe) {
usage(argp, "Invalid worker thread count", 1);
}
}
final Executor executor = Executors.newCachedThreadPool();
final NioServerBossPool boss_pool =
new NioServerBossPool(executor, 1, new Threads.BossThreadNamer());
final NioWorkerPool worker_pool = new NioWorkerPool(executor,
workers, new Threads.WorkerThreadNamer());
factory = new NioServerSocketChannelFactory(boss_pool, worker_pool);
} else {
factory = new OioServerSocketChannelFactory(
Executors.newCachedThreadPool(), Executors.newCachedThreadPool(),
new Threads.PrependThreadNamer());
}
...
try {
tsdb = new TSDB(config);
if (startup != null) {
tsdb.setStartupPlugin(startup);
}
tsdb.initializePlugins(true);
if (config.getBoolean("tsd.storage.hbase.prefetch_meta")) {
tsdb.preFetchHBaseMeta();
}
// Make sure we don't even start if we can't find our tables.
tsdb.checkNecessaryTablesExist().joinUninterruptibly();
registerShutdownHook();
final ServerBootstrap server = new ServerBootstrap(factory);
// This manager is capable of lazy init, but we force an init
// here to fail fast.
final RpcManager manager = RpcManager.instance(tsdb);
server.setPipelineFactory(new PipelineFactory(tsdb, manager, connections_limit));
if (config.hasProperty("tsd.network.backlog")) {
server.setOption("backlog", config.getInt("tsd.network.backlog"));
}
server.setOption("child.tcpNoDelay",
config.getBoolean("tsd.network.tcp_no_delay"));
server.setOption("child.keepAlive",
config.getBoolean("tsd.network.keep_alive"));
server.setOption("reuseAddress",
config.getBoolean("tsd.network.reuse_address"));
// null is interpreted as the wildcard address.
InetAddress bindAddress = null;
if (config.hasProperty("tsd.network.bind")) {
bindAddress = InetAddress.getByName(config.getString("tsd.network.bind"));
}
// we validated the network port config earlier
final InetSocketAddress addr = new InetSocketAddress(bindAddress,
config.getInt("tsd.network.port"));
server.bind(addr);
if (startup != null) {
startup.setReady(tsdb);
}
log.info("Ready to serve on " + addr);
} catch (Throwable e) {
factory.releaseExternalResources();
try {
if (tsdb != null)
tsdb.shutdown().joinUninterruptibly();
} catch (Exception e2) {
log.error("Failed to shutdown HBase client", e2);
}
throw new RuntimeException("Initialization failed", e);
}
// The server is now running in separate threads, we can exit main.
}启动主要分三个过程：
1、加载配置文件
2、初始化netty服务
3、初始化TSDB对象
首先，加载配置文件过程计较简单，就是读取opentsdb.conf配置文件并转换成Properties对象。初始化netty服务也较简单，就是生成一个http服务，接收http请求，主要查看下TSDB对象的初始化。
public TSDB(final HBaseClient client, final Config config) {
this.config = config;
if (client == null) {
final org.hbase.async.Config async_config;
if (config.configLocation() != null && !config.configLocation().isEmpty()) {
try {
async_config = new org.hbase.async.Config(config.configLocation());
} catch (final IOException e) {
throw new RuntimeException("Failed to read the config file: " +
config.configLocation(), e);
}
} else {
async_config = new org.hbase.async.Config();
}
async_config.overrideConfig("hbase.zookeeper.znode.parent",
config.getString("tsd.storage.hbase.zk_basedir"));
async_config.overrideConfig("hbase.zookeeper.quorum",
config.getString("tsd.storage.hbase.zk_quorum"));
this.client = new HBaseClient(async_config);
} else {
this.client = client;
}
String string_mode = config.getString("tsd.mode");
if (Strings.isNullOrEmpty(string_mode)) {
mode = OperationMode.READWRITE;
} else if (string_mode.toLowerCase().equals("ro")) {
mode = OperationMode.READONLY;
} else if (string_mode.toLowerCase().equals("wo")) {
mode = OperationMode.WRITEONLY;
} else {
mode = OperationMode.READWRITE;
}
// SALT AND UID WIDTHS
// Users really wanted this to be set via config instead of having to
// compile. Hopefully they know NOT to change these after writing data.
if (config.hasProperty("tsd.storage.uid.width.metric")) {
METRICS_WIDTH = config.getShort("tsd.storage.uid.width.metric");
}
if (config.hasProperty("tsd.storage.uid.width.tagk")) {
TAG_NAME_WIDTH = config.getShort("tsd.storage.uid.width.tagk");
}
if (config.hasProperty("tsd.storage.uid.width.tagv")) {
TAG_VALUE_WIDTH = config.getShort("tsd.storage.uid.width.tagv");
}
if (config.hasProperty("tsd.storage.max_tags")) {
Const.setMaxNumTags(config.getShort("tsd.storage.max_tags"));
}
if (config.hasProperty("tsd.storage.salt.buckets")) {
Const.setSaltBuckets(config.getInt("tsd.storage.salt.buckets"));
}
if (config.hasProperty("tsd.storage.salt.width")) {
Const.setSaltWidth(config.getInt("tsd.storage.salt.width"));
}
table = config.getString("tsd.storage.hbase.data_table").getBytes(CHARSET);
uidtable = config.getString("tsd.storage.hbase.uid_table").getBytes(CHARSET);
treetable = config.getString("tsd.storage.hbase.tree_table").getBytes(CHARSET);
meta_table = config.getString("tsd.storage.hbase.meta_table").getBytes(CHARSET);
if (config.getBoolean("tsd.core.uid.random_metrics")) {
metrics = new UniqueId(this, uidtable, METRICS_QUAL, METRICS_WIDTH, true);
} else {
metrics = new UniqueId(this, uidtable, METRICS_QUAL, METRICS_WIDTH, false);
}
tag_names = new UniqueId(this, uidtable, TAG_NAME_QUAL, TAG_NAME_WIDTH, false);
tag_values = new UniqueId(this, uidtable, TAG_VALUE_QUAL, TAG_VALUE_WIDTH, false);
compactionq = new CompactionQueue(this);
if (config.hasProperty("tsd.core.timezone")) {
DateTime.setDefaultTimezone(config.getString("tsd.core.timezone"));
}
timer = Threads.newTimer("TSDB Timer");
if (config.getBoolean("tsd.rollups.enable")) {
String conf = config.getString("tsd.rollups.config");
if (Strings.isNullOrEmpty(conf)) {
throw new IllegalArgumentException("Rollups were enabled but "
+ "'tsd.rollups.config' is null or empty.");
}
if (conf.endsWith(".json")) {
try {
conf = Files.toString(new File(conf), Const.UTF8_CHARSET);
} catch (IOException e) {
throw new IllegalArgumentException("Failed to open conf file: "
+ conf, e);
}
}
rollup_config = JSON.parseToObject(conf, RollupConfig.class);
RollupInterval config_default = null;
for (final RollupInterval interval: rollup_config.getRollups().values()) {
if (interval.isDefaultInterval()) {
config_default = interval;
break;
}
}
if (config_default == null) {
throw new IllegalArgumentException("None of the rollup intervals were "
+ "marked as the \"default\".");
}
default_interval = config_default;
tag_raw_data = config.getBoolean("tsd.rollups.tag_raw");
agg_tag_key = config.getString("tsd.rollups.agg_tag_key");
raw_agg_tag_value = config.getString("tsd.rollups.raw_agg_tag_value");
rollups_block_derived = config.getBoolean("tsd.rollups.block_derived");
} else {
rollup_config = null;
default_interval = null;
tag_raw_data = false;
agg_tag_key = null;
raw_agg_tag_value = null;
rollups_block_derived = false;
}
QueryStats.setEnableDuplicates(
config.getBoolean("tsd.query.allow_simultaneous_duplicates"));
if (config.getBoolean("tsd.core.preload_uid_cache")) {
final ByteMap<UniqueId> uid_cache_map = new ByteMap<UniqueId>();
uid_cache_map.put(METRICS_QUAL.getBytes(CHARSET), metrics);
uid_cache_map.put(TAG_NAME_QUAL.getBytes(CHARSET), tag_names);
uid_cache_map.put(TAG_VALUE_QUAL.getBytes(CHARSET), tag_values);
UniqueId.preloadUidCache(this, uid_cache_map);
}
if (config.getString("tsd.core.tag.allow_specialchars") != null) {
Tags.setAllowSpecialChars(config.getString("tsd.core.tag.allow_specialchars"));
}
query_limits = new QueryLimitOverride(this);
// load up the functions that require the TSDB object
ExpressionFactory.addTSDBFunctions(this);
// set any extra tags from the config for stats
StatsCollector.setGlobalTags(config);
LOG.debug(config.dumpConfiguration());
}1、初始化Hbase客户端
if (client == null) {
final org.hbase.async.Config async_config;
if (config.configLocation() != null && !config.configLocation().isEmpty()) {
try {
async_config = new org.hbase.async.Config(config.configLocation());
} catch (final IOException e) {
throw new RuntimeException("Failed to read the config file: " +
config.configLocation(), e);
}
} else {
async_config = new org.hbase.async.Config();
}
async_config.overrideConfig("hbase.zookeeper.znode.parent",
config.getString("tsd.storage.hbase.zk_basedir"));
async_config.overrideConfig("hbase.zookeeper.quorum",
config.getString("tsd.storage.hbase.zk_quorum"));
this.client = new HBaseClient(async_config);
}从配置文件中获取Hbase的相关信息，如zookeeper中Hbase的根目录，之后生成HBaseClient客户端，达到与Hbase交互的目的。从上述可以看出，跟Hbase交互是异步。
2、设置opentsdb实例读写权限，目前有三种权限：只读、只写以及读写。目前实例的权限模式是读写。
String string_mode = config.getString("tsd.mode");
if (Strings.isNullOrEmpty(string_mode)) {
mode = OperationMode.READWRITE;
} else if (string_mode.toLowerCase().equals("ro")) {
mode = OperationMode.READONLY;
} else if (string_mode.toLowerCase().equals("wo")) {
mode = OperationMode.WRITEONLY;
} else {
mode = OperationMode.READWRITE;
}3、设置Hbase表的名称
table = config.getString("tsd.storage.hbase.data_table").getBytes(CHARSET);
uidtable = config.getString("tsd.storage.hbase.uid_table").getBytes(CHARSET);
treetable = config.getString("tsd.storage.hbase.tree_table").getBytes(CHARSET);
meta_table = config.getString("tsd.storage.hbase.meta_table").getBytes(CHARSET);4、设置metrics等参数
if (config.getBoolean("tsd.core.uid.random_metrics")) {
metrics = new UniqueId(this, uidtable, METRICS_QUAL, METRICS_WIDTH, true);
} else {
metrics = new UniqueId(this, uidtable, METRICS_QUAL, METRICS_WIDTH, false);
}
tag_names = new UniqueId(this, uidtable, TAG_NAME_QUAL, TAG_NAME_WIDTH, false);
tag_values = new UniqueId(this, uidtable, TAG_VALUE_QUAL, TAG_VALUE_WIDTH, false);
compactionq = new CompactionQueue(this);5、加载函数
ExpressionFactory.addTSDBFunctions(this);
至此，TSDB对象初始化完毕。
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/69946831/viewspace-3152670/ ，如需转载，请注明出处，否则将追究法律责任。
请登录后发表评论
登录
全部评论
