Golang expvar.Func类(方法)实例源码

func StartDebugServer(address string, sink *lager.ReconfigurableSink, metrics Metrics) (ifrit.Process, error) {
	expvar.Publish("numCPUS", expvar.Func(func() interface{} {
		return metrics.NumCPU()
	}))

	expvar.Publish("numGoRoutines", expvar.Func(func() interface{} {
		return metrics.NumGoroutine()
	}))

	expvar.Publish("loopDevices", expvar.Func(func() interface{} {
		return metrics.LoopDevices()
	}))

	expvar.Publish("backingStores", expvar.Func(func() interface{} {
		return metrics.BackingStores()
	}))

	expvar.Publish("depotDirs", expvar.Func(func() interface{} {
		return metrics.DepotDirs()
	}))

	server := http_server.New(address, handler(sink))
	p := ifrit.Invoke(server)
	select {
	case <-p.Ready():
	case err := <-p.Wait():
		return nil, err
	}
	return p, nil
}

/*
 * Initializations
 */
func init() {
	flag.Parse()
	status.InputEventCount = expvar.NewInt("input_event_count")
	status.OutputEventCount = expvar.NewInt("output_event_count")
	status.ErrorCount = expvar.NewInt("error_count")

	expvar.Publish("connection_status",
		expvar.Func(func() interface{} {
			res := make(map[string]interface{}, 0)
			res["last_connect_time"] = status.LastConnectTime
			res["last_error_text"] = status.LastConnectError
			res["last_error_time"] = status.ErrorTime
			if status.IsConnected {
				res["connected"] = true
				res["uptime"] = time.Now().Sub(status.LastConnectTime).Seconds()
			} else {
				res["connected"] = false
				res["uptime"] = 0.0
			}

			return res
		}))
	expvar.Publish("uptime", expvar.Func(func() interface{} {
		return time.Now().Sub(status.StartTime).Seconds()
	}))
	expvar.Publish("subscribed_events", expvar.Func(func() interface{} {
		return config.EventTypes
	}))

	results = make(chan string, 100)
	output_errors = make(chan error)

	status.StartTime = time.Now()
}

func newOriginAllower(blockedDomains []string, hostname string, gclog logClient, ns *expvar.Map) *originAllower {
	mu := &sync.RWMutex{}
	topKAllDomains := topk.New(100)
	topKOfflistDomains := topk.New(100)
	lifetime := new(expvar.Map).Init()
	ns.Set("lifetime", lifetime)
	lifetime.Set("top_all_domains", expvar.Func(func() interface{} {
		mu.RLock()
		defer mu.RUnlock()
		return topKAllDomains.Keys()
	}))
	lifetime.Set("top_offlist_domains", expvar.Func(func() interface{} {
		mu.RLock()
		defer mu.RUnlock()
		return topKOfflistDomains.Keys()
	}))

	oa := &originAllower{
		m:                  make(map[string]struct{}),
		ns:                 ns,
		hostname:           hostname,
		gclog:              gclog,
		mu:                 mu,
		topKAllDomains:     topKAllDomains,
		topKOfflistDomains: topKOfflistDomains,
	}
	for _, d := range blockedDomains {
		oa.m[d] = struct{}{}
	}
	return oa
}

func init() {
	besticon.SetCacheMaxSize(128)

	expvar.Publish("cacheBytes", expvar.Func(func() interface{} { return besticon.GetCacheStats().Bytes }))
	expvar.Publish("cacheItems", expvar.Func(func() interface{} { return besticon.GetCacheStats().Items }))
	expvar.Publish("cacheGets", expvar.Func(func() interface{} { return besticon.GetCacheStats().Gets }))
	expvar.Publish("cacheHits", expvar.Func(func() interface{} { return besticon.GetCacheStats().Hits }))
	expvar.Publish("cacheEvictions", expvar.Func(func() interface{} { return besticon.GetCacheStats().Evictions }))
}

func initDebug(mux *http.ServeMux) {
	stats.prev = time.Now()

	expvar.Publish("dcp", expvar.Func(dcp))
	expvar.Publish("goroutines", expvar.Func(goroutines))

	mux.Handle("/debug/vars/", http.HandlerFunc(expvarHandler))
	mux.Handle("/debug/pprof/", http.HandlerFunc(pprof.Index))
	mux.Handle("/debug/pprof/cmdline", http.HandlerFunc(pprof.Cmdline))
	mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
	mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
	mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
}

func main() {
	flag.Parse()

	if err := loadContext(); err != nil {
		panic(err)
	}

	ctx := context()
	expvar.Publish("config", &ctx.Config)

	expvar.Publish("codemanager", expvar.Func(func() interface{} {
		return context().InputManager
	}))
	expvar.Publish("queues", expvar.Func(func() interface{} {
		return context().QueueManager
	}))
	expvar.Publish("inflight_runs", expvar.Func(func() interface{} {
		return context().InflightMonitor
	}))
	cachePath := path.Join(ctx.Config.Grader.RuntimePath, "cache")
	go ctx.InputManager.PreloadInputs(
		cachePath,
		grader.NewGraderCachedInputFactory(cachePath),
		&sync.Mutex{},
	)

	// Database
	db, err := sql.Open(
		ctx.Config.Db.Driver,
		ctx.Config.Db.DataSourceName,
	)
	if err != nil {
		panic(err)
	}
	if err := db.Ping(); err != nil {
		panic(err)
	}

	setupMetrics(ctx)
	ctx.Log.Info("omegaUp grader started")

	mux := http.DefaultServeMux
	if ctx.Config.Grader.V1.Enabled {
		registerV1CompatHandlers(mux, db)
		go common.RunServer(&ctx.Config.TLS, mux, ctx.Config.Grader.V1.Port, *insecure)
		mux = http.NewServeMux()
	}

	registerHandlers(mux, db)
	common.RunServer(&ctx.Config.TLS, mux, ctx.Config.Grader.Port, *insecure)
}

func init() {
	expvar.Publish("snmp_request_cache", expvar.Func(func() interface{} {
		requests_mutex.Lock()
		size := requests_cache.Size()
		requests_mutex.Unlock()
		return size
	}))

	expvar.Publish("snmp_bytes_cache", expvar.Func(func() interface{} {
		bytes_mutex.Lock()
		size := bytes_cache.Size()
		bytes_mutex.Unlock()
		return size
	}))
}

func NewRPCExt(name string, maxIdleConnections int, clientBuilder ClientBuilder) *RPCExt {
	r := &RPCExt{
		name:               name,
		clients:            make([]*rpc.Client, 0, maxIdleConnections),
		maxIdleConnections: maxIdleConnections,
		clientBuilder:      clientBuilder,
		closed:             false,

		statRequests:               metrics.NewCounter(),
		statEstablishedConnections: metrics.NewCounter(),
		statLiveConnections:        metrics.NewCounter(),
	}

	m := expvar.NewMap(name + "-rpc")
	m.Set("requests", r.statRequests)
	m.Set("connections.established", r.statEstablishedConnections)
	m.Set("connections.inuse", r.statLiveConnections)
	m.Set("connections.idle", expvar.Func(func() interface{} {
		r.mu.RLock()
		n := len(r.clients)
		r.mu.RUnlock()
		return n
	}))

	return r
}

func init() {
	registry := expvar.Get("registry")
	if registry == nil {
		registry = expvar.NewMap("registry")
	}

	cache := registry.(*expvar.Map).Get("cache")
	if cache == nil {
		cache = &expvar.Map{}
		cache.(*expvar.Map).Init()
		registry.(*expvar.Map).Set("cache", cache)
	}

	storage := cache.(*expvar.Map).Get("storage")
	if storage == nil {
		storage = &expvar.Map{}
		storage.(*expvar.Map).Init()
		cache.(*expvar.Map).Set("storage", storage)
	}

	storage.(*expvar.Map).Set("blobdescriptor", expvar.Func(func() interface{} {
		// no need for synchronous access: the increments are atomic and
		// during reading, we don't care if the data is up to date. The
		// numbers will always *eventually* be reported correctly.
		return blobStatterCacheMetrics
	}))
}

func main() {
	// Initialise our configuration from flags

	var nodeName = flag.String("name", REQUIRED, "Node network name and port, e.g. localhost:3000")
	var gobName = flag.String("gob", "", "Alternative gob network name and port for clients, allowing clients to connect over a different physical interface to nodes.")
	var httpName = flag.String("http", "", "Network name and port for the http ExpVar to listen on.")
	var cborName = flag.String("cbor", "", "Network name and port for the CBOR RPC interface to listen on.")
	var nodePath = flag.String("path", REQUIRED, "Node root path for configuration and log files")
	var clusterID = flag.String("id", "", "Cluster ID that this node is part of")

	flag.Parse()

	if flag.Lookup("help") != nil || flag.Lookup("h") != nil {
		flag.PrintDefaults()
		return
	}

	if *nodeName == REQUIRED {
		log.Printf("name missing.\n")
		flag.PrintDefaults()
		return
	}

	if *nodePath == REQUIRED {
		log.Printf("path missing.\n")
		flag.PrintDefaults()
		return
	}

	// Create our server
	serverNode, err := server.NewServerNode(*nodeName, *gobName, *httpName, *cborName, *nodePath, *clusterID)

	if err != nil {
		log.Fatalf("Unable to start server due to errors.\n")
	}

	expvar.Publish("node", expvar.Func(serverNode.ExpVar))

	//dataLog := &memlog.MemoryLog{}

	// Start a listener to handle incoming requests from other peers
	err = serverNode.ListenConnections()
	if err != nil {
		log.Fatalf("Error starting listener: %v\n", err)
	}

	// Setup signal handling to catch interrupts.
	sigc := make(chan os.Signal, 1)
	signal.Notify(sigc,
		os.Interrupt,
		os.Kill)
	go func() {
		<-sigc
		serverNode.RequestShutdown("Request to terminate process detected.")
	}()

	serverNode.WaitForShutdown()

}

// StartSelfMonitor starts http server on random port and exports expvars.
//
// It tries 1024 ports, starting from startPort and registers some expvars if ok.
func StartSelfMonitor() (string, error) {
	for port := startPort; port < startPort+1024; port++ {
		bind := fmt.Sprintf("localhost:%d", port)
		l, err := net.Listen("tcp", bind)
		if err != nil {
			continue
		}
		l.Close()

		expvar.Publish("Goroutines", expvar.Func(goroutines))
		expvar.Publish("Uptime", expvar.Func(uptime))
		go http.ListenAndServe(bind, nil)
		return bind, nil
	}

	return "", fmt.Errorf("no free ports found")
}

func init() {
	expvar.NewString("goVersion").Set(runtime.Version())
	expvar.NewString("iconVersion").Set(besticon.VersionString)

	expvar.NewString("timeLastDeploy").Set(parseUnixTimeStamp(os.Getenv("DEPLOYED_AT")).String())
	expvar.NewString("timeStartup").Set(time.Now().String())
	expvar.Publish("timeCurrent", expvar.Func(func() interface{} { return time.Now() }))
}

func init() {
	flag.Usage = func() {
		fmt.Fprintln(os.Stderr, "Usage of inbucket [options] <conf file>:")
		flag.PrintDefaults()
	}

	expvar.Publish("uptime", expvar.Func(uptime))
}

func init() {
	raft.SetLogger(capnslog.NewPackageLogger("github.com/coreos/etcd", "raft"))
	expvar.Publish("raft.status", expvar.Func(func() interface{} {
		raftStatusMu.Lock()
		defer raftStatusMu.Unlock()
		return raftStatus()
	}))
}

func initExpvars() {
	expvar.Publish("__binary_path", expvar.Func(func() interface{} { return &binaryPath }))
	expvar.Publish("__config_path", expvar.Func(func() interface{} { return &configPath }))
	expvar.Publish("_command_line", expvar.Func(func() interface{} { return &commandLine }))
	expvar.Publish("_version", expvar.Func(func() interface{} { return VersionInfo }))
	expvar.Publish("_hostname", expvar.Func(func() interface{} { return hostname }))

	expvar.Publish("_config", expvar.Func(func() interface{} {
		cf, err := structs.BadooStripSecretFields(config) // makes a copy of `config`
		if err != nil {
			return struct {
				Error string `json:"error"`
			}{err.Error()}
		}
		return cf
	}))

	expvar.Publish("_service-stats", expvar.Func(func() interface{} {
		stats, err := GatherServiceStats()
		if err != nil {
			return struct {
				Error string `json:"error"`
			}{err.Error()}
		}
		return stats
	}))
}

func publishExtraVars() {
	// By using sync.Once instead of an init() function, we don't clutter
	// the app's expvar export unnecessarily, or risk colliding with it.
	publishOnce.Do(func() {
		expvar.Publish("Goroutines", expvar.Func(func() interface{} {
			return runtime.NumGoroutine()
		}))
	})
}

func Run(address string, sink *lager.ReconfigurableSink) (ifrit.Process, error) {
	expvar.Publish("numCPUS", expvar.Func(func() interface{} {
		return int64(runtime.NumCPU())
	}))

	expvar.Publish("numGoRoutines", expvar.Func(func() interface{} {
		return int64(runtime.NumGoroutine())
	}))

	server := http_server.New(address, handler(sink))
	p := ifrit.Invoke(server)
	select {
	case <-p.Ready():
	case err := <-p.Wait():
		return nil, err
	}
	return p, nil
}

func init() {
	expvar.Publish("kaman.metrics", expvar.Func(func() interface{} {
		counters := Snapshot()
		return map[string]interface{}{
			"Counters": counters,
		}
	}))

}

func ListenAndServe(addr string) tsdb.Chan {
	ch := make(chan *tsdb.Point, MaxQueue)
	s := listen(addr)
	statQueue.Set("", expvar.Func(func() interface{} {
		return len(ch)
	}))
	go s.loop(ch)
	return ch
}

func init() {
	rm := expvar.NewMap("retention")
	rm.Set("SecondsSinceScanCompleted", expvar.Func(secondsSinceRetentionScanCompleted))
	rm.Set("DeletesHist", expRetentionDeletesHist)
	rm.Set("DeletesTotal", expRetentionDeletesTotal)
	rm.Set("Period", expRetentionPeriod)
	rm.Set("RetainedHist", expRetainedHist)
	rm.Set("RetainedCurrent", expRetainedCurrent)
}