From fb2f9fa10a97114b2fd7b9e308b0a218bf7f6578 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ferdinand=20M=C3=BCtsch?= Date: Fri, 11 Oct 2019 09:06:34 +0200 Subject: [PATCH] Ability to schedule summary creation. --- config.ini | 3 +++ go.mod | 1 + go.sum | 8 +++++++ main.go | 11 +++++----- models/config.go | 1 + models/heartbeat.go | 3 +-- services/aggregation.go | 48 ++++++++++++++++++++--------------------- services/summary.go | 3 --- 8 files changed, 43 insertions(+), 35 deletions(-) diff --git a/config.ini b/config.ini index 4cf8e96..28238d5 100644 --- a/config.ini +++ b/config.ini @@ -2,5 +2,8 @@ listen = 127.0.0.1 port = 3000 +[database] +max_connections = 2 + [languages] vue = Vue \ No newline at end of file diff --git a/go.mod b/go.mod index 638e3fd..0dabad3 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.13 require ( github.com/codegangsta/negroni v1.0.0 github.com/gorilla/mux v1.7.3 + github.com/jasonlvhit/gocron v0.0.0-20191007145845-57f89394836a github.com/jinzhu/gorm v1.9.11 github.com/joho/godotenv v1.3.0 github.com/patrickmn/go-cache v2.1.0+incompatible diff --git a/go.sum b/go.sum index a6b1514..bae4be8 100644 --- a/go.sum +++ b/go.sum @@ -11,6 +11,7 @@ github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/codegangsta/negroni v1.0.0 h1:+aYywywx4bnKXWvoWtRfJ91vC59NbEhEY03sZjQhbVY= github.com/codegangsta/negroni v1.0.0/go.mod h1:v0y3T5G7Y1UlFfyxFn/QLRU4a2EuNau2iZY63YTKWo0= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/denisenkom/go-mssqldb v0.0.0-20190515213511-eb9f6a1743f3/go.mod h1:zAg7JM8CkOJ43xKXIj7eRO9kmWm/TW578qo+oDO6tuM= github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs= @@ -20,6 +21,7 @@ github.com/erikstmartin/go-testdb v0.0.0-20160219214506-8d10e4a1bae5/go.mod h1:a github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE= +github.com/go-redis/redis v6.15.5+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA= github.com/go-sql-driver/mysql v1.4.1 h1:g24URVg0OFbNUTx9qqY1IRZ9D9z3iPyi5zKhQZpNwpA= github.com/go-sql-driver/mysql v1.4.1/go.mod h1:zAC/RDZ24gD3HViQzih4MyKcchzm+sOG5ZlKdlhCg5w= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= @@ -41,6 +43,8 @@ github.com/gorilla/mux v1.7.3 h1:gnP5JzjVOuiZD07fKKToCAOjS0yOpj/qPETTXCCS6hw= github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= +github.com/jasonlvhit/gocron v0.0.0-20191007145845-57f89394836a h1:RE/1K1Rht0M9/NoPiGKNRtnzDeG+2usuL/KJOJ9AN6I= +github.com/jasonlvhit/gocron v0.0.0-20191007145845-57f89394836a/go.mod h1:1nXLkt6gXojCECs34KL3+LlZ3gTpZlkPUA8ejW3WeP0= github.com/jinzhu/gorm v1.9.11 h1:gaHGvE+UnWGlbWG4Y3FUwY1EcZ5n6S9WtqBA/uySMLE= github.com/jinzhu/gorm v1.9.11/go.mod h1:bu/pK8szGZ2puuErfU0RwyeNdsf3e6nCX/noXaVxkfw= github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= @@ -59,7 +63,9 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= +github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY= +github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY= github.com/openzipkin/zipkin-go v0.1.6/go.mod h1:QgAqvLzwWbR/WpD4A3cGpPtJrZXNIiJc5AZX7/PBEpw= github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc= github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ= @@ -79,8 +85,10 @@ github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU= github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0= github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/t-tiger/gorm-bulk-insert v0.0.0-20191006042854-7b3d108c15e2 h1:udtWxHsGAV9usmS0mcSKrDv/Mntbg8lJ2KCuCTvXNfQ= github.com/t-tiger/gorm-bulk-insert v0.0.0-20191006042854-7b3d108c15e2/go.mod h1:SK1RZT4TR1aMUNGtbk6YxTPgx2D/gfbxB571QGnAV+c= go.opencensus.io v0.20.1/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk= diff --git a/main.go b/main.go index de26540..e670280 100644 --- a/main.go +++ b/main.go @@ -3,7 +3,6 @@ package main import ( "crypto/md5" "encoding/hex" - "fmt" "log" "net/http" "os" @@ -43,9 +42,10 @@ func readConfig() *models.Config { cfg, err := ini.Load("config.ini") if err != nil { - log.Fatal(fmt.Sprintf("Fail to read file: %v", err)) + log.Fatalf("Fail to read file: %v", err) } + dbMaxConn := cfg.Section("database").Key("max_connections").MustUint(1) addr := cfg.Section("server").Key("listen").MustString("127.0.0.1") port, err := strconv.Atoi(os.Getenv("PORT")) if err != nil { @@ -67,6 +67,7 @@ func readConfig() *models.Config { DbPassword: dbPassword, DbName: dbName, DbDialect: "mysql", + DbMaxConn: dbMaxConn, CustomLanguages: customLangs, } } @@ -103,10 +104,10 @@ func main() { heartbeatSrvc := &services.HeartbeatService{config, db} userSrvc := &services.UserService{config, db} summarySrvc := &services.SummaryService{config, db, heartbeatSrvc, aliasSrvc} - _ = &services.AggregationService{config, db, userSrvc, summarySrvc, heartbeatSrvc} + aggregationSrvc := &services.AggregationService{config, db, userSrvc, summarySrvc, heartbeatSrvc} - // DEBUG ONLY !!! - //aggregationSrvc.Start(time.Second) + // Aggregate heartbeats to summaries and persist them + go aggregationSrvc.Schedule() // Handlers heartbeatHandler := &routes.HeartbeatHandler{HeartbeatSrvc: heartbeatSrvc} diff --git a/models/config.go b/models/config.go index dd61204..9e91fd7 100644 --- a/models/config.go +++ b/models/config.go @@ -8,5 +8,6 @@ type Config struct { DbPassword string DbName string DbDialect string + DbMaxConn uint CustomLanguages map[string]string } diff --git a/models/heartbeat.go b/models/heartbeat.go index a04609e..c714190 100644 --- a/models/heartbeat.go +++ b/models/heartbeat.go @@ -3,7 +3,6 @@ package models import ( "database/sql/driver" "errors" - "fmt" "regexp" "strconv" "strings" @@ -70,7 +69,7 @@ func (j *HeartbeatReqTime) Scan(value interface{}) error { *j = HeartbeatReqTime(value.(time.Time)) break default: - return errors.New(fmt.Sprintf("Unsupported type")) + return errors.New("Unsupported type") } return nil } diff --git a/services/aggregation.go b/services/aggregation.go index e858186..58db77d 100644 --- a/services/aggregation.go +++ b/services/aggregation.go @@ -1,29 +1,17 @@ -/* - << WORK IN PROGRESS >> - Don't use theses classes, yet. - - This aims to implement https://github.com/n1try/wakapi/issues/1. - Idea is to have regularly running, cron-like background jobs that request a summary - from SummaryService for a pre-defined time interval, e.g. 24 hours. Those are persisted - to the database. Once a user request a summary for a certain time frame that partilly - overlaps with pre-generated summaries, those will be aggregated together with actual heartbeats - for the non-overlapping time frames left and right. -*/ - package services import ( "log" + "runtime" "time" + "github.com/jasonlvhit/gocron" "github.com/jinzhu/gorm" "github.com/n1try/wakapi/models" ) const ( - summaryInterval time.Duration = 24 * time.Hour - nSummaryWorkers int = 8 - nPersistWorkers int = 8 + summaryInterval time.Duration = 24 * time.Hour // TODO: Make configurable ) type AggregationService struct { @@ -40,27 +28,32 @@ type AggregationJob struct { To time.Time } -// Use https://godoc.org/github.com/jasonlvhit/gocron to trigger jobs on a regular basis. -func (srv *AggregationService) Start(interval time.Duration) { +// Schedule a job to (re-)generate summaries every day shortly after midnight +// TODO: Make configurable +func (srv *AggregationService) Schedule() { jobs := make(chan *AggregationJob) summaries := make(chan *models.Summary) + defer close(jobs) + defer close(summaries) - for i := 0; i < nSummaryWorkers; i++ { + for i := 0; i < runtime.NumCPU(); i++ { go srv.summaryWorker(jobs, summaries) } - for i := 0; i < nPersistWorkers; i++ { + for i := 0; i < int(srv.Config.DbMaxConn); i++ { go srv.persistWorker(summaries) } - srv.generateJobs(jobs) + gocron.Every(1).Day().At("02:15").Do(srv.trigger, jobs) + <-gocron.Start() } func (srv *AggregationService) summaryWorker(jobs <-chan *AggregationJob, summaries chan<- *models.Summary) { for job := range jobs { if summary, err := srv.SummaryService.Construct(job.From, job.To, &models.User{ID: job.UserID}); err != nil { - log.Printf("Failed to generate summary (%v, %v, %s) – %v.", job.From, job.To, job.UserID, err) + log.Printf("Failed to generate summary (%v, %v, %s) – %v.\n", job.From, job.To, job.UserID, err) } else { + log.Printf("Successfully generated summary (%v, %v, %s).\n", job.From, job.To, job.UserID) summaries <- summary } } @@ -69,19 +62,23 @@ func (srv *AggregationService) summaryWorker(jobs <-chan *AggregationJob, summar func (srv *AggregationService) persistWorker(summaries <-chan *models.Summary) { for summary := range summaries { if err := srv.SummaryService.Insert(summary); err != nil { - log.Printf("Failed to save summary (%v, %v, %s) – %v.", summary.UserID, summary.FromTime, summary.ToTime, err) + log.Printf("Failed to save summary (%v, %v, %s) – %v.\n", summary.UserID, summary.FromTime, summary.ToTime, err) } } } -func (srv *AggregationService) generateJobs(jobs chan<- *AggregationJob) error { +func (srv *AggregationService) trigger(jobs chan<- *AggregationJob) error { + log.Println("Generating summaries.") + users, err := srv.UserService.GetAll() if err != nil { + log.Println(err) return err } latestSummaries, err := srv.SummaryService.GetLatestByUser() if err != nil { + log.Println(err) return err } @@ -99,6 +96,7 @@ func (srv *AggregationService) generateJobs(jobs chan<- *AggregationJob) error { firstHeartbeats, err := srv.HeartbeatService.GetFirstUserHeartbeats(missingUserIDs) if err != nil { + log.Println(err) return err } @@ -120,12 +118,12 @@ func generateUserJobs(userId string, lastAggregation time.Time, jobs chan<- *Agg if lastAggregation.Hour() == 0 { from = lastAggregation } else { - nextDay := lastAggregation.Add(24 * time.Hour) + nextDay := lastAggregation.Add(summaryInterval) from = time.Date(nextDay.Year(), nextDay.Month(), nextDay.Day(), 0, 0, 0, 0, lastAggregation.Location()) } for from.Before(end) && to.Before(end) { - to = from.Add(24 * time.Hour) + to = from.Add(summaryInterval) jobs <- &AggregationJob{userId, from, to} from = to } diff --git a/services/summary.go b/services/summary.go index 6b19aa9..91c8db4 100644 --- a/services/summary.go +++ b/services/summary.go @@ -2,7 +2,6 @@ package services import ( "errors" - "fmt" "math" "sort" "time" @@ -93,7 +92,6 @@ func (srv *SummaryService) Construct(from, to time.Time, user *models.User) (*mo } func (srv *SummaryService) Insert(summary *models.Summary) error { - fmt.Println("Saving summary", summary) if err := srv.Db.Create(summary).Error; err != nil { return err } @@ -238,7 +236,6 @@ func mergeSummaries(summaries []*models.Summary) (*models.Summary, error) { maxTime = s.ToTime } - // TODO: Multi-thread ? finalSummary.Projects = mergeSummaryItems(finalSummary.Projects, s.Projects) finalSummary.Languages = mergeSummaryItems(finalSummary.Languages, s.Languages) finalSummary.Editors = mergeSummaryItems(finalSummary.Editors, s.Editors)