Function startDataNode

cmd/aresd/cmd/cmd.go:256–299

start datanode in distributed mode

(cfg common.AresServerConfig, logger common.Logger, scope tally.Scope, httpWrappers ...utils.HTTPHandlerWrapper)

// start datanode in distributed mode
func startDataNode(cfg common.AresServerConfig, logger common.Logger, scope tally.Scope, httpWrappers ...utils.HTTPHandlerWrapper) {
	opts := datanode.NewOptions().SetServerConfig(cfg).SetInstrumentOptions(utils.NewOptions()).SetBootstrapOptions(bootstrap.NewOptions()).SetHTTPWrappers(httpWrappers)

	var topo topology.Topology
	etcdCfg := cfg.Cluster.Etcd
	etcdCfg.Service = utils.DataNodeServiceName(cfg.Cluster.Namespace)
	configServiceCli, err := etcdCfg.NewClient(instrument.NewOptions())
	if err != nil {
		logger.With("error", err.Error()).Fatal("failed to create etcd client", err)
	}

	txnStore, err := configServiceCli.Txn()
	if err != nil {
		logger.With("error", err.Error()).Fatal("failed to create txn store")
	}

	enumReader := etcd.NewEnumMutator(txnStore, etcd.NewTableSchemaMutator(txnStore, zap.NewExample().Sugar()))

	dynamicOptions := topology.NewDynamicOptions().
		SetConfigServiceClient(configServiceCli).
		SetQueryOptions(services.NewQueryOptions().SetIncludeUnhealthy(true)).
		SetServiceID(services.NewServiceID().
			SetZone(etcdCfg.Zone).
			SetName(etcdCfg.Service).
			SetEnvironment(etcdCfg.Env))
	topo, err = topology.NewDynamicInitializer(dynamicOptions).Init()
	if err != nil {
		logger.Fatal("Failed to initialize dynamic topology,", err)
	}

	dataNode, err := datanode.NewDataNode(cfg.Cluster.InstanceID, topo, enumReader, opts)
	if err != nil {
		logger.Fatal("Failed to create datanode,", err)
	}
	defer dataNode.Close()

	// preparing
	err = dataNode.Open()
	if err != nil {
		logger.Fatal("Failed to open datanode,", err)
	}
	// start serving traffic
	dataNode.Serve()
}