Feat/store processed stats in pgsql database

.gitignore

@@ -1,3 +1,6 @@
+# Editor settings
+.vscode/
+
 # Env files
 .env
 

benchttp/benchttp.go → benchttp/report.go

@@ -2,6 +2,7 @@ package benchttp
 
 // Report represents the result of a Benchttp benchmark run.
 type Report struct {
+	ID        string
 	Benchmark Benchmark
 }
 

benchttp/service.go

@@ -0,0 +1,8 @@
+package benchttp
+
+// InsertService defines the interface to implement by a
+// data layer facade.
+type InsertionService interface {
+	// Create stores stats in database.
+	Insert(Stats) error
+}

benchttp/stats.go

@@ -0,0 +1,33 @@
+package benchttp
+
+import (
+	"time"
+
+	"github.com/benchttp/worker/stats"
+)
+
+// StatsDescriptor contains a computed stats group description information.
+type StatsDescriptor struct {
+	ID         string    `json:"id"`
+	UserID     string    `json:"userID"`
+	Tag        string    `json:"tag"`
+	FinishedAt time.Time `json:"finishedAt"`
+}
+
+// Stats contains StatsDescriptor, Codestats and stats.Stats of a given computed stats group.
+type Stats struct {
+	Descriptor StatsDescriptor          `json:"descriptor"`
+	Code       stats.StatusDistribution `json:"code"`
+	Time       stats.Common             `json:"time"`
+}
+
+// Config contains the information needed to connect to the database
+// and set max idle connections and open connections number.
+type Config struct {
+	Host     string
+	User     string
+	Password string
+	DBName   string
+	IdleConn int
+	MaxConn  int
+}

firestoreconv/firestoreconv.go

@@ -26,6 +26,11 @@ var ErrMapValueField = errors.New("key is not in protobuf map value")
 
 // Report converts a Firestore event payload to a usable benchttp.Report.
 func Report(v *firestore.Value) (benchttp.Report, error) {
+	id, ok := v.Fields["id"]
+	if !ok {
+		return benchttp.Report{}, fmt.Errorf(`%w: "%s"`, ErrMapValueField, "id")
+	}
+
 	benchmark, ok := v.Fields["benchmark"]
 	if !ok {
 		return benchttp.Report{}, fmt.Errorf(`%w: "%s"`, ErrMapValueField, "benchmark")
@@ -55,7 +60,7 @@ func Report(v *firestore.Value) (benchttp.Report, error) {
 		}
 	}
 
-	var report benchttp.Report
+	report := benchttp.Report{ID: *id.StringValue}
 	report.Benchmark.Records = records
 
 	return report, nil

firestoreconv/firestoreconv_test.go

@@ -14,6 +14,9 @@ func TestToBenchmark(t *testing.T) {
 	e := firestore.DocumentEventData{
 		Value: &firestore.Value{
 			Fields: map[string]firestore.OldValueField{
+				"id": {
+					StringValue: newString("1234"),
+				},
 				"benchmark": {
 					MapValue: &firestore.MapValue{
 						Fields: map[string]firestore.MapValueField{
@@ -55,6 +58,7 @@ func TestToBenchmark(t *testing.T) {
 	}
 
 	want := benchttp.Report{
+		ID: "1234",
 		Benchmark: benchttp.Benchmark{
 			Records: []benchttp.Record{
 				{Time: 100, Code: 200},
@@ -76,3 +80,8 @@ func TestToBenchmark(t *testing.T) {
 func newInt64(x int64) *int64 {
 	return &x
 }
+
+// newString returns a pointer to the given string value.
+func newString(x string) *string {
+	return &x
+}

go.mod

@@ -3,6 +3,9 @@ module github.com/benchttp/worker
 go 1.16
 
 require (
+	github.com/GoogleCloudPlatform/cloudsql-proxy v1.29.0
 	github.com/googleapis/google-cloudevents-go v0.2.0
+	github.com/joho/godotenv v1.4.0
+	github.com/lib/pq v1.10.4
 	github.com/montanaflynn/stats v0.6.6
 )

main.go

@@ -2,22 +2,37 @@ package worker
 
 import (
 	"context"
+	"errors"
 	"log"
+	"os"
+	"time"
 
 	"github.com/googleapis/google-cloudevents-go/cloud/firestore/v1"
 
+	"github.com/benchttp/worker/benchttp"
 	"github.com/benchttp/worker/firestoreconv"
+	"github.com/benchttp/worker/postgresql"
 	"github.com/benchttp/worker/stats"
 )
 
 // Digest is a Cloud Function triggered by a Firestore create document
 // event to extract, compute and store statistics of a Benchttp run.
+// It also stores the computed data in a SQL database.
 func Digest(ctx context.Context, e firestore.DocumentEventData) error {
 	r, err := firestoreconv.Report(e.Value)
 	if err != nil {
 		return err
 	}
 
+	cfg, err := envConfig()
+	if err != nil {
+		return err
+	}
+	insertionService, err := postgresql.NewInsertionService(cfg)
+	if err != nil {
+		return err
+	}
+
 	codes, times := r.Benchmark.Values()
 
 	timestats, err := stats.ComputeCommon(times)
@@ -34,5 +49,56 @@ func Digest(ctx context.Context, e firestore.DocumentEventData) error {
 
 	log.Printf("codestats: %+v", codestats)
 
+	// TO DO: get user id. Using "1" here for the moment.
+	statsToInsert := buildStats(timestats, codestats, "firestore_id", "1")
+
+	if err := insertionService.Insert(statsToInsert); err != nil {
+		return err
+	}
+
 	return nil
 }
+
+func envConfig() (benchttp.Config, error) {
+	var config benchttp.Config
+
+	config.Host = os.Getenv("PSQL_HOST")
+	if config.Host == "" {
+		return config, errors.New("PSQL_HOST environment variable not found")
+	}
+
+	config.User = os.Getenv("PSQL_USER")
+	if config.User == "" {
+		return config, errors.New("PSQL_USER environment variable not found")
+	}
+
+	config.Password = os.Getenv("PSQL_PASSWORD")
+	if config.Password == "" {
+		return config, errors.New("PSQL_PASSWORD environment variable not found")
+	}
+
+	config.DBName = os.Getenv("PSQL_NAME")
+	if config.DBName == "" {
+		return config, errors.New("PSQL_NAME environment variable not found")
+	}
+
+	config.IdleConn = 10
+	config.MaxConn = 25
+
+	return config, nil
+}
+
+// buildStats builds a benchttp.Stats object.
+// Descriptor.FinishedAt is set at time.now().
+func buildStats(timestats stats.Common, codestats stats.StatusDistribution, reportID, userID string) benchttp.Stats {
+	computedstats := benchttp.Stats{
+		Descriptor: benchttp.StatsDescriptor{
+			ID:         reportID,
+			UserID:     userID,
+			FinishedAt: time.Now(),
+		},
+		Time: timestats,
+		Code: codestats,
+	}
+	return computedstats
+}

postgresql/connection.go

@@ -0,0 +1,41 @@
+package postgresql
+
+import (
+	"database/sql"
+	"fmt"
+
+	_ "github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/dialers/postgres" // blank import
+
+	"github.com/benchttp/worker/benchttp"
+)
+
+// InsertionService implements benchttp.InsertionService interface.
+type InsertionService struct {
+	db *sql.DB
+}
+
+// NewInsertionService connects to the database and provides an InsertionService
+// implementing benchttp.InsertionService, which provides a method to insert
+// data in database.
+func NewInsertionService(config benchttp.Config) (InsertionService, error) {
+	dbURI := fmt.Sprintf("host=%s user=%s password=%s dbname=%s sslmode=disable",
+		config.Host,
+		config.User,
+		config.Password,
+		config.DBName)
+
+	db, err := sql.Open("cloudsqlpostgres", dbURI)
+	if err != nil {
+		return InsertionService{}, err
+	}
+
+	err = db.Ping()
+	if err != nil {
+		return InsertionService{}, err
+	}
+
+	db.SetMaxIdleConns(config.IdleConn)
+	db.SetMaxOpenConns(config.MaxConn)
+
+	return InsertionService{db}, nil
+}

postgresql/stats.go

@@ -0,0 +1,114 @@
+package postgresql
+
+import (
+	"github.com/lib/pq"
+
+	"github.com/benchttp/worker/benchttp"
+)
+
+// Insert inserts computed stats (benchttp.Stats) in the database.
+// nolint:gocognit
+func (s InsertionService) Insert(stats benchttp.Stats) error {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+
+	insertStatsDescriptor, err := tx.Prepare(`
+	INSERT INTO public.stats_descriptor(
+		id,
+		user_id,
+		finished_at)
+		VALUES($1, $2, $3)`)
+	if err != nil {
+		if err := tx.Rollback(); err != nil {
+			return err
+		}
+		return err
+	}
+	defer insertStatsDescriptor.Close()
+
+	if _, err = insertStatsDescriptor.Exec(
+		stats.Descriptor.ID,
+		stats.Descriptor.UserID,
+		stats.Descriptor.FinishedAt,
+	); err != nil {
+		if err := tx.Rollback(); err != nil {
+			return err
+		}
+		return err
+	}
+
+	insertTimestats, err := tx.Prepare(`
+	INSERT INTO public.timestats(
+		stats_descriptor_id, 
+		min, 
+		max, 
+		mean, 
+		median, 
+		standard_deviation, 
+		deciles) 
+	VALUES($1, $2, $3, $4, $5, $6, $7)`)
+	if err != nil {
+		if err := tx.Rollback(); err != nil {
+			return err
+		}
+		return err
+	}
+	defer insertTimestats.Close()
+
+	if _, err = insertTimestats.Exec(
+		stats.Descriptor.ID,
+		stats.Time.Min,
+		stats.Time.Max,
+		stats.Time.Mean,
+		stats.Time.Median,
+		stats.Time.StdDev,
+		pq.Array(stats.Time.Deciles),
+	); err != nil {
+		if err := tx.Rollback(); err != nil {
+			return err
+		}
+		return err
+	}
+
+	insertCodestats, err := tx.Prepare(`
+	INSERT INTO codestats(
+		stats_descriptor_id,
+		code_1xx,
+		code_2xx,
+		code_3xx,
+		code_4xx,
+		code_5xx
+	) VALUES($1, $2, $3, $4, $5, $6)`)
+	if err != nil {
+		if err := tx.Rollback(); err != nil {
+			return err
+		}
+		return err
+	}
+	defer insertCodestats.Close()
+
+	if _, err = insertCodestats.Exec(
+		stats.Descriptor.ID,
+		stats.Code.Status1xx,
+		stats.Code.Status2xx,
+		stats.Code.Status3xx,
+		stats.Code.Status4xx,
+		stats.Code.Status5xx,
+	); err != nil {
+		if err := tx.Rollback(); err != nil {
+			return err
+		}
+		return err
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		if err := tx.Rollback(); err != nil {
+			return err
+		}
+		return err
+	}
+	return nil
+}