2
0
Эх сурвалжийг харах

add log hooks to main package

Jordan Schalm 8 жил өмнө
parent
commit
2ae49516ad

+ 0 - 4
cmd/guerrillad/serve.go

@@ -18,8 +18,6 @@ import (
 
 	"github.com/flashmob/go-guerrilla"
 	"github.com/flashmob/go-guerrilla/backends"
-	"github.com/flashmob/go-guerrilla/dashboard"
-	// "github.com/flashmob/go-guerrilla/dashboard"
 )
 
 var (
@@ -44,8 +42,6 @@ func init() {
 		"/var/run/go-guerrilla.pid", "Path to the pid file")
 
 	rootCmd.AddCommand(serveCmd)
-
-	log.AddHook(dashboard.NewLogHook())
 }
 
 func sigHandler(app guerrilla.Guerrilla) {

+ 8 - 2
config.go

@@ -2,8 +2,14 @@ package guerrilla
 
 // AppConfig is the holder of the configuration of the app
 type AppConfig struct {
-	Servers      []ServerConfig `json:"servers"`
-	AllowedHosts []string       `json:"allowed_hosts"`
+	Dashboard    DashboardConfig `json:"dashboard"`
+	Servers      []ServerConfig  `json:"servers"`
+	AllowedHosts []string        `json:"allowed_hosts"`
+}
+
+type DashboardConfig struct {
+	Enabled         bool   `json:"enabled"`
+	ListenInterface string `json:"listen_interface"`
 }
 
 // ServerConfig specifies config options for a single server

+ 66 - 88
dashboard/dashboard.go

@@ -12,18 +12,15 @@ import (
 
 const (
 	dashboard      = "index.html"
-	login          = "login.html"
 	dashboardPath  = "dashboard/html/index.html"
-	loginPath      = "dashboard/html/login.html"
 	sessionTimeout = time.Hour * 24 // TODO replace with config
 )
 
 var (
 	// Cache of HTML templates
-	templates = template.Must(template.ParseFiles(dashboardPath, loginPath))
+	templates = template.Must(template.ParseFiles(dashboardPath))
 	config    *Config
-	sessions  sessionStore
-	store     *dataStore
+	sessions  map[string]*session
 )
 
 var upgrader = websocket.Upgrader{
@@ -32,93 +29,89 @@ var upgrader = websocket.Upgrader{
 }
 
 type Config struct {
-	Username        string
-	Password        string
 	ListenInterface string
 }
 
 func Run(c *Config) {
 	config = c
+	sessions = map[string]*session{}
 	r := mux.NewRouter()
 	r.HandleFunc("/", indexHandler)
-	r.HandleFunc("/login", loginHandler)
-	r.HandleFunc("/logout", logoutHandler)
 	r.HandleFunc("/ws", webSocketHandler)
 
 	rand.Seed(time.Now().UnixNano())
 
-	sessions = make(sessionStore)
-	go sessions.cleaner(sessionTimeout)
-	store = newDataStore()
-	go ramListener(tickInterval, store)
+	go dataListener(tickInterval)
 
 	http.ListenAndServe(c.ListenInterface, r)
 }
 
 func indexHandler(w http.ResponseWriter, r *http.Request) {
-	if isLoggedIn(r) {
-		w.WriteHeader(http.StatusOK)
-		templates.ExecuteTemplate(w, dashboard, nil)
-	} else {
-		http.Redirect(w, r, "/login", http.StatusTemporaryRedirect)
-	}
-}
-
-func loginHandler(w http.ResponseWriter, r *http.Request) {
-	switch r.Method {
-	case "GET":
-		if isLoggedIn(r) {
-			http.Redirect(w, r, "/", http.StatusTemporaryRedirect)
-		} else {
-			templates.ExecuteTemplate(w, login, nil)
-		}
-
-	case "POST":
-		user := r.FormValue("username")
-		pass := r.FormValue("password")
-
-		if user == config.Username && pass == config.Password {
-			err := startSession(w, r)
-			if err != nil {
-				w.WriteHeader(http.StatusInternalServerError)
-				// TODO Internal error
-				return
-			}
-			http.Redirect(w, r, "/", http.StatusSeeOther)
-		} else {
-			templates.ExecuteTemplate(w, login, nil) // TODO info about failed login
-		}
-
-	default:
-		w.WriteHeader(http.StatusMethodNotAllowed)
+	c, err := r.Cookie("SID")
+	_, sidExists := sessions[c.Value]
+	if err != nil || !sidExists {
+		// No SID cookie
+		startSession(w, r)
 	}
+	w.WriteHeader(http.StatusOK)
+	templates.ExecuteTemplate(w, dashboard, nil)
 }
 
-func logoutHandler(w http.ResponseWriter, r *http.Request) {
-	switch r.Method {
-	case "POST":
-		sess := getSession(r)
-		if sess == nil {
-			w.WriteHeader(http.StatusForbidden)
-			return
-		}
-
-		store.unsubscribe(sess.id)
-		sess.expires = time.Now()
-		http.Redirect(w, r, "/", http.StatusSeeOther)
-
-	default:
-		w.WriteHeader(http.StatusMethodNotAllowed)
-	}
-}
+// func loginHandler(w http.ResponseWriter, r *http.Request) {
+// 	switch r.Method {
+// 	case "GET":
+// 		if isLoggedIn(r) {
+// 			http.Redirect(w, r, "/", http.StatusTemporaryRedirect)
+// 		} else {
+// 			templates.ExecuteTemplate(w, login, nil)
+// 		}
+//
+// 	case "POST":
+// 		user := r.FormValue("username")
+// 		pass := r.FormValue("password")
+//
+// 		if user == config.Username && pass == config.Password {
+// 			err := startSession(w, r)
+// 			if err != nil {
+// 				w.WriteHeader(http.StatusInternalServerError)
+// 				// TODO Internal error
+// 				return
+// 			}
+// 			http.Redirect(w, r, "/", http.StatusSeeOther)
+// 		} else {
+// 			templates.ExecuteTemplate(w, login, nil) // TODO info about failed login
+// 		}
+//
+// 	default:
+// 		w.WriteHeader(http.StatusMethodNotAllowed)
+// 	}
+// }
+//
+// func logoutHandler(w http.ResponseWriter, r *http.Request) {
+// 	switch r.Method {
+// 	case "POST":
+// 		sess := getSession(r)
+// 		if sess == nil {
+// 			w.WriteHeader(http.StatusForbidden)
+// 			return
+// 		}
+//
+// 		store.unsubscribe(sess.id)
+// 		sess.expires = time.Now()
+// 		http.Redirect(w, r, "/", http.StatusSeeOther)
+//
+// 	default:
+// 		w.WriteHeader(http.StatusMethodNotAllowed)
+// 	}
+// }
 
 func webSocketHandler(w http.ResponseWriter, r *http.Request) {
-	if !isLoggedIn(r) {
-		w.WriteHeader(http.StatusForbidden)
+	sess := getSession(r)
+	if sess == nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		// TODO Internal error
 		return
 	}
-
-	sess := getSession(r)
 	conn, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
@@ -126,14 +119,15 @@ func webSocketHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	sess.ws = conn
-	c := make(chan *point)
+	c := make(chan *dataFrame)
 	sess.send = c
+	// TODO send store contents at connection time
 	store.subscribe(sess.id, c)
 	go sess.receive()
 	go sess.transmit()
 }
 
-func startSession(w http.ResponseWriter, r *http.Request) error {
+func startSession(w http.ResponseWriter, r *http.Request) {
 	sessionID := newSessionID()
 
 	cookie := &http.Cookie{
@@ -144,14 +138,11 @@ func startSession(w http.ResponseWriter, r *http.Request) error {
 	}
 
 	sess := &session{
-		start:   time.Now(),
-		expires: time.Now().Add(sessionTimeout), // TODO config for this
-		id:      sessionID,
+		id: sessionID,
 	}
 
 	http.SetCookie(w, cookie)
 	sessions[sessionID] = sess
-	return nil
 }
 
 func getSession(r *http.Request) *session {
@@ -168,16 +159,3 @@ func getSession(r *http.Request) *session {
 
 	return sess
 }
-
-func isLoggedIn(r *http.Request) bool {
-	sess := getSession(r)
-	if sess == nil {
-		return false
-	}
-
-	if !sess.valid() {
-		return false
-	}
-
-	return true
-}

+ 65 - 51
dashboard/datastore.go

@@ -8,33 +8,53 @@ import (
 )
 
 const (
-	tickInterval = time.Second
+	tickInterval = time.Second * 5
 	maxWindow    = time.Hour * 24
 	maxTicks     = int(maxWindow / tickInterval)
 )
 
+// Log for sending client events from the server to the dashboard.
+var (
+	LogHook = logHook(1)
+	store   = newDataStore()
+)
+
 type dataStore struct {
-	ram  []*point
-	subs map[string]chan<- *point
+	// List of samples of RAM usage
+	ramTicks []point
+	// List of samples of number of connected clients
+	nClientTicks []point
+	// Up-to-date number of clients
+	nClients uint64
+	subs     map[string]chan<- *dataFrame
 }
 
 func newDataStore() *dataStore {
+	subs := make(map[string]chan<- *dataFrame)
 	return &dataStore{
-		ram:  make([]*point, 0, maxTicks),
-		subs: make(map[string]chan<- *point),
+		ramTicks:     make([]point, 0, maxTicks),
+		nClientTicks: make([]point, 0, maxTicks),
+		subs:         subs,
 	}
 }
 
-func (ds *dataStore) addPoint(p *point) {
-	if len(ds.ram) == int(maxTicks) {
-		ds.ram = append(ds.ram[1:], p)
+func (ds *dataStore) addRAMPoint(p point) {
+	if len(ds.ramTicks) == int(maxTicks) {
+		ds.ramTicks = append(ds.ramTicks[1:], p)
 	} else {
-		ds.ram = append(ds.ram, p)
+		ds.ramTicks = append(ds.ramTicks, p)
 	}
-	ds.notify(p)
 }
 
-func (ds *dataStore) subscribe(id string, c chan<- *point) {
+func (ds *dataStore) addNClientPoint(p point) {
+	if len(ds.nClientTicks) == int(maxTicks) {
+		ds.nClientTicks = append(ds.nClientTicks[1:], p)
+	} else {
+		ds.nClientTicks = append(ds.nClientTicks, p)
+	}
+}
+
+func (ds *dataStore) subscribe(id string, c chan<- *dataFrame) {
 	ds.subs[id] = c
 }
 
@@ -42,7 +62,7 @@ func (ds *dataStore) unsubscribe(id string) {
 	delete(ds.subs, id)
 }
 
-func (ds *dataStore) notify(p *point) {
+func (ds *dataStore) notify(p *dataFrame) {
 	for _, c := range ds.subs {
 		select {
 		case c <- p:
@@ -56,59 +76,53 @@ type point struct {
 	Y uint64    `json:"y"`
 }
 
-func ramListener(interval time.Duration, store *dataStore) {
+func dataListener(interval time.Duration) {
 	ticker := time.Tick(interval)
 	memStats := &runtime.MemStats{}
 
 	for {
 		t := <-ticker
 		runtime.ReadMemStats(memStats)
-		store.addPoint(&point{t, memStats.Alloc})
+		ramPoint := point{t, memStats.Alloc}
+		nClientPoint := point{t, store.nClients}
+		log.Info("datastore:89", ramPoint, nClientPoint)
+		store.addRAMPoint(ramPoint)
+		store.addNClientPoint(nClientPoint)
+		store.notify(&dataFrame{
+			Ram:      ramPoint,
+			NClients: nClientPoint,
+		})
 	}
 }
 
-type SendEvent struct {
-	timeStamp     time.Time
-	helo          string
-	remoteAddress string
+type dataFrame struct {
+	Ram      point `json:"ram"`
+	NClients point `json:"n_clients"`
+	// top5Helo []string // TODO add for aggregation
+	// top5IP   []string
 }
 
-type LogHook struct {
-	events chan *SendEvent
-}
+type logHook int
 
-func NewLogHook() *LogHook {
-	events := make(chan *SendEvent)
-	return &LogHook{events}
-}
-
-func (h *LogHook) Levels() []log.Level {
+func (h logHook) Levels() []log.Level {
 	return []log.Level{log.InfoLevel}
 }
 
-func (h *LogHook) Fire(e *log.Entry) error {
-	// helo, ok := e.Data["helo"]
-	// if !ok {
-	// 	return nil
-	// }
-	// heloStr, ok := helo.(string)
-	// if !ok {
-	// 	return nil
-	// }
-	//
-	// addr, ok := e.Data["remoteAddress"]
-	// if !ok {
-	// 	return nil
-	// }
-	// addrStr, ok := addr.(string)
-	// if !ok {
-	// 	return nil
-	// }
-	//
-	// h.events <- &SendEvent{
-	// 	timeStamp:     e.Time,
-	// 	helo:          heloStr,
-	// 	remoteAddress: addrStr,
-	// }
+func (h logHook) Fire(e *log.Entry) error {
+	event, ok := e.Data["event"]
+	if !ok {
+		return nil
+	}
+	event, ok = event.(string)
+	if !ok {
+		return nil
+	}
+
+	switch event {
+	case "connect":
+		store.nClients++
+	case "disconnect":
+		store.nClients--
+	}
 	return nil
 }

+ 0 - 2
dashboard/exe/main.go

@@ -6,8 +6,6 @@ import (
 
 func main() {
 	dashboard.Run(&dashboard.Config{
-		Password:        "password",
-		Username:        "admin",
 		ListenInterface: ":8080",
 	})
 }

+ 28 - 15
dashboard/html/index.html

@@ -6,24 +6,37 @@
 	<body>
 		<canvas id="ram-graph" width="500" height="200"></canvas>
 	</body>
-	<script src="https://cdnjs.cloudflare.com/ajax/libs/smoothie/1.27.0/smoothie.min.js"></script>
-	<script src="https://cdnjs.cloudflare.com/ajax/libs/Chart.js/2.4.0/Chart.min.js"></script>
+	<!--script src="https://cdnjs.cloudflare.com/ajax/libs/smoothie/1.27.0/smoothie.min.js"></script-->
+	<script src="https://cdnjs.cloudflare.com/ajax/libs/moment.js/2.17.1/moment.min.js"></script>
+	<!--script src="https://cdnjs.cloudflare.com/ajax/libs/Chart.js/2.4.0/Chart.min.js"></script-->
+
+	<!-- imports global MG variable -->
+	<script src="https://cdnjs.cloudflare.com/ajax/libs/jquery/3.1.1/jquery.min.js"></script>
+	<script src="https://cdnjs.cloudflare.com/ajax/libs/d3/4.1.1/d3.min.js"></script>
+	<script src="https://cdnjs.cloudflare.com/ajax/libs/metrics-graphics/2.11.0/metricsgraphics.js"></script>
 	<script>
-	var conn = new WebSocket('ws://localhost:8080/ws');
-	var smoothie = new SmoothieChart();
-	var ram = new TimeSeries();
 
-	smoothie.addTimeSeries(ram);
-	smoothie.streamTo(document.getElementById('ram-graph'), 1000);
 
-	conn.onclose = function(event) {
-		console.log(event);
-	}
+	var data = [1,2,3,4,5,6,7,8,9,10].map(n => ({date: moment().add(n, 'days').format('YYYY-MM-DD'), value: n}));
+	console.log(data);
+	MG.data_graphic({
+		title: "Line Chart",
+		description: "This is a simple line chart. You can remove the area portion by adding area: false to the arguments list.",
+		data: data,
+		width: 600,
+		height: 200,
+		right: 40,
+		target: document.getElementById('ram-graph'),
+		x_accessor: 'date',
+		y_accessor: 'value'
+	});
+
+	// var smoothie = new SmoothieChart();
+	// var ram = new TimeSeries();
+	//
+	// smoothie.addTimeSeries(ram);
+	// smoothie.streamTo(document.getElementById('ram-graph'), 1000);
+
 
-	conn.onmessage = function(event) {
-		console.log(JSON.parse(event.data));
-		var point = JSON.parse(event.data);
-		ram.append(new Date(point.t).getTime(), point.y);
-	}
 	</script>
 </html>

+ 16 - 37
dashboard/session.go

@@ -1,6 +1,7 @@
 package dashboard
 
 import (
+	"encoding/json"
 	"math/rand"
 	"time"
 
@@ -17,18 +18,12 @@ const (
 
 var idCharset = []byte("qwertyuiopasdfghjklzxcvbnmQWERTYUIOPASDFGHJKLZXCVBNM1234567890")
 
+// Represents an active session with a client
 type session struct {
-	start, expires time.Time
-	id             string
-	// Whether we have a valid
-	alive bool
-	ws    *websocket.Conn
+	id string
+	ws *websocket.Conn
 	// Messages to send over the websocket are received on this channel
-	send <-chan *point
-}
-
-func (s *session) valid() bool {
-	return s.expires.After(time.Now())
+	send <-chan *dataFrame
 }
 
 // Receives messages from the websocket connection associated with a session
@@ -59,52 +54,36 @@ func (s *session) transmit() {
 	defer s.ws.Close()
 	defer ticker.Stop()
 
+	// Label for loop to allow breaking from within switch statement
+transmit:
 	for {
 		select {
 		case p, ok := <-s.send:
+			data, err := json.Marshal(p)
+			log.Info("session:61", string(data), err)
+			log.Info("session:59", p.NClients, p.Ram)
 			s.ws.SetWriteDeadline(time.Now().Add(writeWait))
-			if !ok || !s.valid() {
+			if !ok {
 				s.ws.WriteMessage(websocket.CloseMessage, []byte{})
-				break
+				break transmit
 			}
 
-			err := s.ws.WriteJSON(p)
+			err = s.ws.WriteJSON(p)
+			log.Info("session:67", err)
 			if err != nil {
 				log.WithError(err).Debug("Failed to write next websocket message. Closing connection")
-				break
+				break transmit
 			}
 		case <-ticker.C:
 			s.ws.SetWriteDeadline(time.Now().Add(writeWait))
 			if err := s.ws.WriteMessage(websocket.PingMessage, []byte{}); err != nil {
 				log.WithError(err).Debug("Failed to write next websocket message. Closing connection")
-				break
+				break transmit
 			}
 		}
 	}
 }
 
-type sessionStore map[string]*session
-
-// Remove expired sessions
-func (ss sessionStore) clean() {
-	now := time.Now()
-	for id, sess := range ss {
-		if sess.expires.Before(now) {
-			delete(ss, id)
-		}
-	}
-}
-
-// Cleans the store on each tick
-func (ss sessionStore) cleaner(interval time.Duration) {
-	ticker := time.NewTicker(interval)
-	defer ticker.Stop()
-	for {
-		<-ticker.C
-		ss.clean()
-	}
-}
-
 // Returns a random alphanumeric 10-character ID
 func newSessionID() string {
 	mask := int64(63)

+ 13 - 1
guerrilla.go

@@ -2,10 +2,16 @@ package guerrilla
 
 import (
 	"errors"
-	log "github.com/Sirupsen/logrus"
 	"sync"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/flashmob/go-guerrilla/dashboard"
 )
 
+func init() {
+	log.AddHook(dashboard.LogHook)
+}
+
 type Guerrilla interface {
 	Start() (startErrors []error)
 	Shutdown()
@@ -58,6 +64,12 @@ func (g *guerrilla) Start() (startErrors []error) {
 	// wait for all servers to start
 	startWG.Wait()
 
+	if g.Config.Dashboard.Enabled {
+		go dashboard.Run(&dashboard.Config{
+			ListenInterface: g.Config.Dashboard.ListenInterface,
+		})
+	}
+
 	// close, then read any errors
 	close(errs)
 	for err := range errs {

+ 12 - 2
server.go

@@ -164,6 +164,12 @@ func (server *server) upgradeToTLS(client *client) bool {
 func (server *server) closeConn(client *client) {
 	client.conn.Close()
 	client.conn = nil
+	log.WithFields(map[string]interface{}{
+		"event":   "disconnect",
+		"address": client.RemoteAddress,
+		"helo":    client.Helo,
+		"id":      client.ID,
+	}).Info("Close client")
 }
 
 // Reads from the client until a terminating sequence is encountered,
@@ -224,8 +230,12 @@ func (server *server) isShuttingDown() bool {
 // Handles an entire client SMTP exchange
 func (server *server) handleClient(client *client) {
 	defer server.closeConn(client)
-
-	log.Infof("Handle client [%s], id: %d", client.RemoteAddress, client.ID)
+	log.WithFields(map[string]interface{}{
+		"event":   "connect",
+		"address": client.RemoteAddress,
+		"helo":    client.Helo,
+		"id":      client.ID,
+	}).Info("Handle client")
 
 	// Initial greeting
 	greeting := fmt.Sprintf("220 %s SMTP Guerrilla(%s) #%d (%d) %s gr:%d",