Method Serve

datanode/datanode.go:471–515

()

}

func (d *dataNode) Serve() {
	// wait for server is ready to serve
	<-d.readyCh

	// start advertising to the cluster
	d.advertise()
	// enable archiving jobs
	if !d.opts.ServerConfig().SchedulerOff {
		d.memStore.GetScheduler().EnableJobType(memCom.ArchivingJobType, true)
		d.opts.InstrumentOptions().Logger().Info("archiving jobs enabled")
	}

	// start server
	router := mux.NewRouter()
	httpWrappers := append([]utils.HTTPHandlerWrapper{utils.WithMetricsFunc}, d.opts.HTTPWrappers()...)
	schemaRouter := router.PathPrefix("/schema")
	if d.opts.ServerConfig().Cluster.Enable {
		schemaRouter = schemaRouter.Methods(http.MethodGet)
	}

	d.handlers.schemaHandler.Register(schemaRouter.Subrouter(), httpWrappers...)
	d.handlers.enumHandler.Register(router.PathPrefix("/schema").Subrouter(), httpWrappers...)
	d.handlers.dataHandler.Register(router.PathPrefix("/data").Subrouter(), httpWrappers...)
	d.handlers.queryHandler.Register(router.PathPrefix("/query").Subrouter(), httpWrappers...)

	router.PathPrefix("/swagger/").Handler(d.handlers.swaggerHandler)
	router.PathPrefix("/node_modules/").Handler(d.handlers.nodeModuleHandler)
	router.HandleFunc("/health", utils.WithMetricsFunc(d.handlers.healthCheckHandler.HealthCheck))
	router.HandleFunc("/version", d.handlers.healthCheckHandler.Version)

	// Support CORS calls.
	allowOrigins := handlers.AllowedOrigins([]string{"*"})
	allowHeaders := handlers.AllowedHeaders([]string{"Accept", "Accept-Language", "Content-Language", "Origin", "Content-Type"})
	allowMethods := handlers.AllowedMethods([]string{"GET", "PUT", "POST", "DELETE", "OPTIONS"})

	// record time from data node started to actually serving
	d.metrics.restartTimer.Record(utils.Now().Sub(d.startedAt))

	// start batch reporter
	batchStatsReporter := memstore.NewBatchStatsReporter(5*60, d.memStore, d)
	go batchStatsReporter.Run()

	d.opts.InstrumentOptions().Logger().Infof("Starting HTTP server on port %d with max connection %d", d.opts.ServerConfig().Port, d.opts.ServerConfig().HTTP.MaxConnections)
	utils.LimitServe(d.opts.ServerConfig().Port, handlers.CORS(allowOrigins, allowHeaders, allowMethods)(mixedHandler(d.grpcServer, router)), d.opts.ServerConfig().HTTP)
}

func (d *dataNode) advertise() {
	serviceID := services.NewServiceID().