From 663bec4fd78c3e1cb4865bcf01b5e322bc326b27 Mon Sep 17 00:00:00 2001
From: Jan Semmelink <jan@uafrica.com>
Date: Wed, 10 Nov 2021 11:49:18 +0200
Subject: [PATCH] Restructure mainly queues to separate consumer from producer,
because consumer is a service, but service must be able to produce, which
caused cyclic dependencies. Now audit and logs can use a producer to send SQS
events, and service can use logs and audits.
---
api/README.md | 4 +-
api/api.go | 20 +-
api/lambda.go | 12 +-
audit/audit.go | 8 -
audit/{event.go => change.go} | 48 +++-
audit/file_audit.go | 65 -----
audit/no_audit.go | 20 --
consumer/README.md | 4 +
consumer/check.go | 7 +
{queues => consumer}/consumer.go | 4 +-
{queues => consumer}/context.go | 11 +-
{queues => consumer}/handler.go | 2 +-
.../mem => consumer/mem_consumer}/README.md | 0
.../mem => consumer/mem_consumer}/consumer.go | 52 ++--
.../mem => consumer/mem_consumer}/producer.go | 24 +-
{queues => consumer}/router.go | 2 +-
.../sqs => consumer/sqs_consumer}/consumer.go | 86 ++++---
cron/cron.go | 9 +-
cron/handler.go | 2 +-
examples/core/api/main.go | 2 -
examples/core/app/users/users.go | 4 +-
examples/core/cron/main.go | 1 -
examples/core/db/database.go | 4 +-
examples/core/email/notify.go | 6 +-
examples/core/sqs/main.go | 4 +-
logs/README.md | 43 ++++
logs/api-logs.go | 230 ++++++++++++++++++
logs/cron.logs.go | 3 +
logs/sqs-logs.go | 96 ++++++++
queues/audit.go | 55 -----
queues/check.go | 5 -
{service => queues}/event.go | 15 +-
{service => queues}/producer.go | 4 +-
queues/{sqs => sqs_producer}/README.md | 0
queues/{sqs => sqs_producer}/producer.go | 14 +-
service/README.md | 11 +-
service/context.go | 22 +-
service/service.go | 33 +--
service/start.go | 2 +-
39 files changed, 583 insertions(+), 351 deletions(-)
delete mode 100644 audit/audit.go
rename audit/{event.go => change.go} (79%)
delete mode 100644 audit/file_audit.go
delete mode 100644 audit/no_audit.go
create mode 100644 consumer/README.md
create mode 100644 consumer/check.go
rename {queues => consumer}/consumer.go (70%)
rename {queues => consumer}/context.go (84%)
rename {queues => consumer}/handler.go (98%)
rename {queues/mem => consumer/mem_consumer}/README.md (100%)
rename {queues/mem => consumer/mem_consumer}/consumer.go (76%)
rename {queues/mem => consumer/mem_consumer}/producer.go (54%)
rename {queues => consumer}/router.go (99%)
rename {queues/sqs => consumer/sqs_consumer}/consumer.go (67%)
create mode 100644 logs/README.md
create mode 100644 logs/api-logs.go
create mode 100644 logs/cron.logs.go
create mode 100644 logs/sqs-logs.go
delete mode 100644 queues/audit.go
delete mode 100644 queues/check.go
rename {service => queues}/event.go (91%)
rename {service => queues}/producer.go (55%)
rename queues/{sqs => sqs_producer}/README.md (100%)
rename queues/{sqs => sqs_producer}/producer.go (90%)
diff --git a/api/README.md b/api/README.md
index a464c77..1435a49 100644
--- a/api/README.md
+++ b/api/README.md
@@ -204,9 +204,7 @@ API documentation will eventually be generated from doc tags in your structs. Th
# API Logger
-API uses the go-utils/logger at the moment. Eventually it will be possible to use other loggers, and customise api logs. Audits can already be customised using ```api.WithAuditor()```.
-
-By default, go-utils/logger writes JSON records. At the start of the example api main function the logger is configured to write console format and log at DEBUG level... more to come on this front too including triggers for full debug in production on selected handlers or specific events, and logging different levels for selected code packages to avoid log clutter without having to remove debug from code.
+API Logs are written from global variables using logs.LogIncomingAPIRequest()
# Router Path Parameters
diff --git a/api/api.go b/api/api.go
index 219cfa7..5cd5e16 100644
--- a/api/api.go
+++ b/api/api.go
@@ -6,10 +6,10 @@ import (
"runtime/debug"
"github.com/aws/aws-lambda-go/lambda"
- "gitlab.com/uafrica/go-utils/audit"
+ "gitlab.com/uafrica/go-utils/consumer/mem_consumer"
"gitlab.com/uafrica/go-utils/errors"
- queues_mem "gitlab.com/uafrica/go-utils/queues/mem"
- queues_sqs "gitlab.com/uafrica/go-utils/queues/sqs"
+ "gitlab.com/uafrica/go-utils/queues"
+ "gitlab.com/uafrica/go-utils/queues/sqs_producer"
"gitlab.com/uafrica/go-utils/service"
"gitlab.com/uafrica/go-utils/string_utils"
)
@@ -52,19 +52,13 @@ type Api struct {
}
//wrap Service.WithStarter to return api, else cannot be chained
-func (api Api) WithStarter(name string, starter service.IStarter) Api {
+func (api Api) WithStarter(name string, starter service.Starter) Api {
api.Service = api.Service.WithStarter(name, starter)
return api
}
//wrap else cannot be chained
-func (api Api) WithAuditor(auditor audit.Auditor) Api {
- api.Service = api.Service.WithAuditor(auditor)
- return api
-}
-
-//wrap else cannot be chained
-func (api Api) WithProducer(producer service.Producer) Api {
+func (api Api) WithProducer(producer queues.Producer) Api {
api.Service = api.Service.WithProducer(producer)
return api
}
@@ -126,11 +120,11 @@ func (api Api) Run() {
if (os.Getenv("LOG_LEVEL") == "debug") && api.localQueueEventHandlers != nil {
//use in-memory channels for async events
api.Debugf("Using in-memory channels for async events ...")
- api = api.WithProducer(queues_mem.NewProducer(queues_mem.NewConsumer(api.Service, api.localQueueEventHandlers)))
+ api = api.WithProducer(mem_consumer.NewProducer(mem_consumer.New(api.Service, api.localQueueEventHandlers)))
} else {
//use SQS for async events
api.Debugf("Using SQS queue producer for async events ...")
- api = api.WithProducer(queues_sqs.NewProducer(api.requestIDHeaderKey))
+ api = api.WithProducer(sqs_producer.New(api.requestIDHeaderKey))
}
//run as an AWS Lambda function
diff --git a/api/lambda.go b/api/lambda.go
index 3d2479e..a36bfbf 100644
--- a/api/lambda.go
+++ b/api/lambda.go
@@ -14,6 +14,7 @@ import (
"github.com/aws/aws-lambda-go/lambdacontext"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/logger"
+ "gitlab.com/uafrica/go-utils/logs"
)
func (api Api) NewContext(baseCtx context.Context, requestID string, request events.APIGatewayProxyRequest) (Context, error) {
@@ -57,7 +58,6 @@ func (api Api) Handler(baseCtx context.Context, apiGatewayProxyReq events.APIGat
if err != nil {
return res, err
}
-
//report handler crashes
if api.crashReporter != nil {
@@ -99,14 +99,8 @@ func (api Api) Handler(baseCtx context.Context, apiGatewayProxyReq events.APIGat
if api.requestIDHeaderKey != "" {
res.Headers[api.requestIDHeaderKey] = Ctx.RequestID()
}
- if err := api.Service.WriteValues(Ctx.StartTime(), time.Now(), Ctx.RequestID(), map[string]interface{}{
- "direction": "incoming",
- "type": "api",
- "request_id": Ctx.RequestID(),
- "request": Ctx.Request(),
- "response": res},
- ); err != nil {
- Ctx.Errorf("failed to audit: %+v", err)
+ if err := logs.LogIncomingAPIRequest(Ctx.StartTime(), apiGatewayProxyReq, res); err != nil {
+ Ctx.Errorf("failed to log: %+v", err)
}
}()
diff --git a/audit/audit.go b/audit/audit.go
deleted file mode 100644
index 2fdf2bf..0000000
--- a/audit/audit.go
+++ /dev/null
@@ -1,8 +0,0 @@
-package audit
-
-import "time"
-
-type Auditor interface {
- WriteValues(startTime, endTime time.Time, requestID string, values map[string]interface{}) error
- WriteEvent(requestID string, event Event) error
-}
diff --git a/audit/event.go b/audit/change.go
similarity index 79%
rename from audit/event.go
rename to audit/change.go
index dc61743..e1d9649 100644
--- a/audit/event.go
+++ b/audit/change.go
@@ -9,26 +9,60 @@ import (
"time"
"github.com/r3labs/diff/v2"
+ "gitlab.com/uafrica/go-utils/errors"
+ "gitlab.com/uafrica/go-utils/queues"
"gitlab.com/uafrica/go-utils/reflection"
)
-type Event struct {
+var producer queues.Producer
+
+func Init(p queues.Producer) {
+ producer = p
+}
+
+func SaveDataChange(
+ requestID string,
+ source string,
+ eventType string,
+ orgValue interface{},
+ newValue interface{},
+) error {
+ if producer == nil {
+ return errors.Errorf("audit queue producer not set")
+ }
+
+ changeRecord, err := NewChangeRecord(source, eventType, orgValue, newValue)
+ if err != nil {
+ return errors.Wrapf(err, "fail to determine changes")
+ }
+ if _, err := producer.NewEvent("AUDIT").
+ Type("audit").
+ RequestID(requestID).
+ Send(changeRecord); err != nil {
+ return errors.Wrapf(err, "failed to send data change record")
+ }
+ return nil
+}
+
+type ChangeRecord struct {
ID int64 `json:"id"`
ObjectID string `json:"object_id"`
Type string `json:"type"`
Source string `json:"source"`
Timestamp time.Time `json:"timestamp"`
- Change map[string]interface{} `json:"change"`
+ Changes map[string]interface{} `json:"changes"`
}
+//purpose:
+// Creates a record describing a change of data
//parameters:
// source could be "" then defaults to "SYSTEM" or specify the user name that made the change
// orgValue and newValue could be nil
-// they are compared and changes are logged
-func NewEvent(source string, eventType string, orgValue, newValue interface{}) (Event, error) {
+// they are compared and changes are recorded
+func NewChangeRecord(source string, eventType string, orgValue, newValue interface{}) (ChangeRecord, error) {
changelog, err := diff.Diff(orgValue, newValue)
if err != nil {
- return Event{}, err
+ return ChangeRecord{}, err
}
changes := map[string]interface{}{}
@@ -129,12 +163,12 @@ func NewEvent(source string, eventType string, orgValue, newValue interface{}) (
objectIDString = getStringValue(newValue, "Key")
}
- event := Event{
+ event := ChangeRecord{
ObjectID: objectIDString,
Source: source,
Type: eventType,
Timestamp: time.Now(),
- Change: changes,
+ Changes: changes,
}
if event.Source == "" {
event.Source = "SYSTEM"
diff --git a/audit/file_audit.go b/audit/file_audit.go
deleted file mode 100644
index 7d340e1..0000000
--- a/audit/file_audit.go
+++ /dev/null
@@ -1,65 +0,0 @@
-package audit
-
-import (
- "encoding/json"
- "os"
- "time"
-
- "gitlab.com/uafrica/go-utils/errors"
-)
-
-//creates auditor that writes to file, which could be os.Stderr or os.Stdout for debugging
-func File(f *os.File) Auditor {
- if f == nil {
- panic(errors.Errorf("cannot create file auditor with f=nil"))
- }
- return &fileAudit{
- f: f,
- }
-}
-
-type fileAudit struct {
- f *os.File
-}
-
-func (fa fileAudit) WriteValues(startTime, endTime time.Time, requestID string, values map[string]interface{}) error {
- if fa.f == nil {
- return errors.Errorf("auditor is closed")
- }
- obj := map[string]interface{}{
- "start_time": startTime,
- "end_time": endTime,
- "duration": endTime.Sub(startTime),
- "request_id": requestID,
- "values": values,
- }
- jsonObj, err := json.Marshal(obj)
- if err != nil {
- return errors.Wrapf(err, "failed to JSON encode audit values")
- }
- if _, err := fa.f.Write(jsonObj); err != nil {
- return errors.Wrapf(err, "failed to write audit values to file")
- }
- return nil
-}
-
-func (fa fileAudit) WriteEvent(requestID string, event Event) error {
- if fa.f == nil {
- return errors.Errorf("auditor is closed")
- }
- obj := map[string]interface{}{
- "start_time": event.Timestamp,
- "end_time": event.Timestamp,
- "duration": 0,
- "request_id": requestID,
- "values": event,
- }
- jsonObj, err := json.Marshal(obj)
- if err != nil {
- return errors.Wrapf(err, "failed to JSON encode audit event")
- }
- if _, err := fa.f.Write(jsonObj); err != nil {
- return errors.Wrapf(err, "failed to write audit event to file")
- }
- return nil
-}
diff --git a/audit/no_audit.go b/audit/no_audit.go
deleted file mode 100644
index 930f413..0000000
--- a/audit/no_audit.go
+++ /dev/null
@@ -1,20 +0,0 @@
-package audit
-
-import (
- "time"
-)
-
-//creates auditor that writes nothiong
-func None() Auditor {
- return noAudit{}
-}
-
-type noAudit struct{}
-
-func (noAudit) WriteValues(startTime, endTime time.Time, requestID string, values map[string]interface{}) error {
- return nil
-}
-
-func (noAudit) WriteEvent(requestID string, event Event) error {
- return nil
-}
diff --git a/consumer/README.md b/consumer/README.md
new file mode 100644
index 0000000..2d8745e
--- /dev/null
+++ b/consumer/README.md
@@ -0,0 +1,4 @@
+# Consumer
+
+Consumes a queue of events in the same way that API processes HTTP requests.
+Consumer is a type of service, just like API and CRON are also types of services.
diff --git a/consumer/check.go b/consumer/check.go
new file mode 100644
index 0000000..3b77c4f
--- /dev/null
+++ b/consumer/check.go
@@ -0,0 +1,7 @@
+package consumer
+
+import "gitlab.com/uafrica/go-utils/service"
+
+type Checker interface {
+ Check(service.Context) (interface{}, error)
+}
diff --git a/queues/consumer.go b/consumer/consumer.go
similarity index 70%
rename from queues/consumer.go
rename to consumer/consumer.go
index ec4c50d..6278ae9 100644
--- a/queues/consumer.go
+++ b/consumer/consumer.go
@@ -1,10 +1,10 @@
-package queues
+package consumer
import "gitlab.com/uafrica/go-utils/service"
//IConsumer is the interface implemented by both mem and sqs consumer
type Consumer interface {
- WithStarter(name string, starter service.IStarter) Consumer
+ WithStarter(name string, starter service.Starter) Consumer
Run()
ProcessFile(filename string) error
}
diff --git a/queues/context.go b/consumer/context.go
similarity index 84%
rename from queues/context.go
rename to consumer/context.go
index c3321fe..ef182c8 100644
--- a/queues/context.go
+++ b/consumer/context.go
@@ -1,4 +1,4 @@
-package queues
+package consumer
import (
"context"
@@ -6,13 +6,14 @@ import (
"reflect"
"gitlab.com/uafrica/go-utils/errors"
+ "gitlab.com/uafrica/go-utils/queues"
"gitlab.com/uafrica/go-utils/service"
)
//Context within a consumer to process an event
type Context interface {
service.Context
- Event() service.Event //the event start started this context in the consumer
+ Event() queues.Event //the event start started this context in the consumer
GetRecord(recordType reflect.Type) (interface{}, error) //extract struct value from event data
}
@@ -20,10 +21,10 @@ var contextInterfaceType = reflect.TypeOf((*Context)(nil)).Elem()
type queuesContext struct {
service.Context
- event service.Event
+ event queues.Event
}
-func NewContext(service service.Service, event service.Event) (Context, error) {
+func NewContext(service service.Service, event queues.Event) (Context, error) {
baseCtx := context.Background()
serviceContext, err := service.NewContext(baseCtx, event.RequestIDValue, map[string]interface{}{
"message_type": event.TypeName,
@@ -39,7 +40,7 @@ func NewContext(service service.Service, event service.Event) (Context, error) {
return ctx, nil
}
-func (ctx queuesContext) Event() service.Event {
+func (ctx queuesContext) Event() queues.Event {
return ctx.event
}
diff --git a/queues/handler.go b/consumer/handler.go
similarity index 98%
rename from queues/handler.go
rename to consumer/handler.go
index 36ba9ec..bc09bd7 100644
--- a/queues/handler.go
+++ b/consumer/handler.go
@@ -1,4 +1,4 @@
-package queues
+package consumer
import (
"reflect"
diff --git a/queues/mem/README.md b/consumer/mem_consumer/README.md
similarity index 100%
rename from queues/mem/README.md
rename to consumer/mem_consumer/README.md
diff --git a/queues/mem/consumer.go b/consumer/mem_consumer/consumer.go
similarity index 76%
rename from queues/mem/consumer.go
rename to consumer/mem_consumer/consumer.go
index 317d1dc..99274c9 100644
--- a/queues/mem/consumer.go
+++ b/consumer/mem_consumer/consumer.go
@@ -1,4 +1,4 @@
-package mem
+package mem_consumer
import (
"encoding/json"
@@ -10,55 +10,49 @@ import (
"time"
"github.com/google/uuid"
- "gitlab.com/uafrica/go-utils/audit"
+ "gitlab.com/uafrica/go-utils/consumer"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/queues"
"gitlab.com/uafrica/go-utils/service"
)
-func NewConsumer(s service.Service, routes map[string]interface{}) queues.Consumer {
+func New(s service.Service, routes map[string]interface{}) consumer.Consumer {
if s == nil {
panic("NewConsumer(service==nil)")
}
- router, err := queues.NewRouter(routes)
+ router, err := consumer.NewRouter(routes)
if err != nil {
panic(fmt.Sprintf("cannot create router: %+v", err))
}
- c := &consumer{
+ c := &memConsumer{
Service: s,
router: router,
queues: map[string]*queue{},
}
//create a producer that will produce into this consumer
- c.producer = &producer{
+ c.producer = &memProducer{
consumer: c,
}
c.Service = c.Service.WithProducer(c.producer)
return c
}
-type consumer struct {
+type memConsumer struct {
sync.Mutex
service.Service
- router queues.Router
- producer *producer
+ router consumer.Router
+ producer queues.Producer
queues map[string]*queue
}
//wrap Service.WithStarter to return cron, else cannot be chained
-func (consumer *consumer) WithStarter(name string, starter service.IStarter) queues.Consumer {
+func (consumer *memConsumer) WithStarter(name string, starter service.Starter) consumer.Consumer {
consumer.Service = consumer.Service.WithStarter(name, starter)
return consumer
}
-//wrap else cannot be chained
-func (consumer *consumer) WithAuditor(auditor audit.Auditor) queues.Consumer {
- consumer.Service = consumer.Service.WithAuditor(auditor)
- return consumer
-}
-
-func (consumer *consumer) Queue(name string) (*queue, error) {
+func (consumer *memConsumer) Queue(name string) (*queue, error) {
consumer.Lock()
defer consumer.Unlock()
q, ok := consumer.queues[name]
@@ -66,7 +60,7 @@ func (consumer *consumer) Queue(name string) (*queue, error) {
q = &queue{
consumer: consumer,
name: name,
- ch: make(chan service.Event),
+ ch: make(chan queues.Event),
}
go q.run()
consumer.queues[name] = q
@@ -76,20 +70,20 @@ func (consumer *consumer) Queue(name string) (*queue, error) {
//do not call this - when using local producer, the consumer is automatically running
//for each queue you send to, and processing from q.run()
-func (consumer *consumer) Run() {
+func (consumer *memConsumer) Run() {
panic(errors.Errorf("DO NOT RUN LOCAL CONSUMER"))
}
-func (consumer *consumer) ProcessFile(filename string) error {
+func (consumer *memConsumer) ProcessFile(filename string) error {
f, err := os.Open(filename)
if err != nil {
return errors.Wrapf(err, "failed to open queue event file %s", filename)
}
defer f.Close()
- var event service.Event
+ var event queues.Event
if err := json.NewDecoder(f).Decode(&event); err != nil {
- return errors.Wrapf(err, "failed to read service.Event from file %s", filename)
+ return errors.Wrapf(err, "failed to read queues.Event from file %s", filename)
}
q := queue{
@@ -107,16 +101,16 @@ func (consumer *consumer) ProcessFile(filename string) error {
}
type queue struct {
- consumer *consumer
+ consumer *memConsumer
name string
- ch chan service.Event
+ ch chan queues.Event
}
func (q *queue) run() {
// logger.Debugf("Q(%s) Start", q.name)
for event := range q.ch {
//process in background because some event processing sends to itself then wait for some responses on new events on the same queue!!!
- go func(event service.Event) {
+ go func(event queues.Event) {
// logger.Debugf("Q(%s) process start: %+v", q.name, event)
err := q.process(event)
if err != nil {
@@ -129,7 +123,7 @@ func (q *queue) run() {
// logger.Debugf("Q(%s) STOPPED", q.name)
}
-func (q *queue) process(event service.Event) error {
+func (q *queue) process(event queues.Event) error {
//todo: create context with logger
rand.Seed(time.Now().Unix())
@@ -137,7 +131,7 @@ func (q *queue) process(event service.Event) error {
// if q.crashReporter != nil {
// defer q.crashReporter.Catch(ctx)
// }
- ctx, err := queues.NewContext(q.consumer.Service, event)
+ ctx, err := consumer.NewContext(q.consumer.Service, event)
if err != nil {
return err
}
@@ -147,7 +141,7 @@ func (q *queue) process(event service.Event) error {
if err != nil {
return errors.Wrapf(err, "unhandled event type(%v)", event.TypeName)
}
- handler, ok := sqsHandler.(queues.Handler)
+ handler, ok := sqsHandler.(consumer.Handler)
if !ok {
return errors.Errorf("messageType(%v) unsupported signature: %T", event.TypeName, sqsHandler)
}
@@ -185,7 +179,7 @@ func (q *queue) process(event service.Event) error {
return nil
} //queue.process()
-func (q *queue) Send(event service.Event) (msgID string, err error) {
+func (q *queue) Send(event queues.Event) (msgID string, err error) {
event.MessageID = uuid.New().String()
q.ch <- event
return event.MessageID, nil
diff --git a/queues/mem/producer.go b/consumer/mem_consumer/producer.go
similarity index 54%
rename from queues/mem/producer.go
rename to consumer/mem_consumer/producer.go
index b32c4c4..f1b51f0 100644
--- a/queues/mem/producer.go
+++ b/consumer/mem_consumer/producer.go
@@ -1,29 +1,29 @@
-package mem
+package mem_consumer
import (
+ "gitlab.com/uafrica/go-utils/consumer"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/logger"
"gitlab.com/uafrica/go-utils/queues"
- "gitlab.com/uafrica/go-utils/service"
)
//can only produce locally if also consuming local
-func NewProducer(memConsumer queues.Consumer) service.Producer {
- if memConsumer == nil {
- panic(errors.Errorf("cannot product locally without consumer"))
+func NewProducer(consumer consumer.Consumer) queues.Producer {
+ if consumer == nil {
+ panic(errors.Errorf("cannot produce local events without mem consumer"))
}
- mc, ok := memConsumer.(*consumer)
+ mc, ok := consumer.(*memConsumer)
if !ok {
- panic(errors.Errorf("NewProducer(%T) is not a mem consumer", memConsumer))
+ panic(errors.Errorf("NewProducer(consumer=%T) is not a mem consumer", consumer))
}
return mc.producer
}
-type producer struct {
- consumer *consumer
+type memProducer struct {
+ consumer *memConsumer
}
-func (producer *producer) Send(event service.Event) (string, error) {
+func (producer *memProducer) Send(event queues.Event) (string, error) {
logger.Debugf("MEM producer.queue(%s) Sending event %+v", event.QueueName, event)
q, err := producer.consumer.Queue(event.QueueName)
if err != nil {
@@ -38,3 +38,7 @@ func (producer *producer) Send(event service.Event) (string, error) {
logger.Debugf("MEM producer.queue(%s) SENT event %+v", event.QueueName, event)
return msgID, nil
}
+
+func (producer *memProducer) NewEvent(queueName string) queues.Event {
+ return queues.NewEvent(producer, queueName)
+}
diff --git a/queues/router.go b/consumer/router.go
similarity index 99%
rename from queues/router.go
rename to consumer/router.go
index f235736..744cf1a 100644
--- a/queues/router.go
+++ b/consumer/router.go
@@ -1,4 +1,4 @@
-package queues
+package consumer
import (
"fmt"
diff --git a/queues/sqs/consumer.go b/consumer/sqs_consumer/consumer.go
similarity index 67%
rename from queues/sqs/consumer.go
rename to consumer/sqs_consumer/consumer.go
index 1c31e5f..bbf2e5b 100644
--- a/queues/sqs/consumer.go
+++ b/consumer/sqs_consumer/consumer.go
@@ -1,4 +1,4 @@
-package sqs
+package sqs_consumer
import (
"context"
@@ -15,18 +15,20 @@ import (
"github.com/aws/aws-lambda-go/lambda"
"github.com/aws/aws-lambda-go/lambdacontext"
"github.com/google/uuid"
- "gitlab.com/uafrica/go-utils/audit"
+ "gitlab.com/uafrica/go-utils/consumer"
"gitlab.com/uafrica/go-utils/errors"
+ "gitlab.com/uafrica/go-utils/logs"
"gitlab.com/uafrica/go-utils/queues"
+ "gitlab.com/uafrica/go-utils/queues/sqs_producer"
"gitlab.com/uafrica/go-utils/service"
)
-func NewConsumer(requestIDHeaderKey string, routes map[string]interface{}) queues.Consumer {
+func New(requestIDHeaderKey string, routes map[string]interface{}) consumer.Consumer {
env := os.Getenv("ENVIRONMENT") //todo: support config loading for local dev and env for lambda in prod
if env == "" {
env = "dev"
}
- router, err := queues.NewRouter(routes)
+ router, err := consumer.NewRouter(routes)
if err != nil {
panic(fmt.Sprintf("cannot create router: %+v", err))
}
@@ -41,43 +43,38 @@ func NewConsumer(requestIDHeaderKey string, routes map[string]interface{}) queue
}
s := service.New().
- WithProducer(NewProducer(requestIDHeaderKey))
- return consumer{
+ WithProducer(sqs_producer.New(requestIDHeaderKey))
+
+ return sqsConsumer{
Service: s,
env: env,
router: router,
requestIDHeaderKey: requestIDHeaderKey,
ConstantMessageType: sqsMessageType,
- checks: map[string]queues.ICheck{},
+ checks: map[string]consumer.Checker{},
}
}
-type consumer struct {
+type sqsConsumer struct {
service.Service
env string
- router queues.Router
+ router consumer.Router
requestIDHeaderKey string
ConstantMessageType string //from os.Getenv("SQS_MESSAGE_TYPE")
- checks map[string]queues.ICheck
+ checks map[string]consumer.Checker
}
//wrap Service.WithStarter to return cron, else cannot be chained
-func (consumer consumer) WithStarter(name string, starter service.IStarter) queues.Consumer {
- consumer.Service = consumer.Service.WithStarter(name, starter)
- return consumer
-}
-
-//wrap else cannot be chained
-func (consumer consumer) WithAuditor(auditor audit.Auditor) queues.Consumer {
- consumer.Service = consumer.Service.WithAuditor(auditor)
- return consumer
+func (c sqsConsumer) WithStarter(name string, starter service.Starter) consumer.Consumer {
+ c.Service = c.Service.WithStarter(name, starter)
+ return c
}
-func (consumer consumer) Run() {
- lambda.Start(consumer.Handler)
+func (c sqsConsumer) Run() {
+ lambda.Start(c.Handler)
}
-func (consumer consumer) ProcessFile(filename string) error {
+func (c sqsConsumer) ProcessFile(filename string) error {
f, err := os.Open(filename)
if err != nil {
return errors.Wrapf(err, "failed to open queue event file %s", filename)
@@ -89,7 +86,7 @@ func (consumer consumer) ProcessFile(filename string) error {
return errors.Wrapf(err, "failed to read sqs event from file %s", filename)
}
- if consumer.Handler(
+ if c.Handler(
lambdacontext.NewContext(
context.Background(),
&lambdacontext.LambdaContext{
@@ -106,7 +103,7 @@ func (consumer consumer) ProcessFile(filename string) error {
return nil
}
-func (consumer consumer) Handler(baseCtx context.Context, lambdaEvent events.SQSEvent) error {
+func (c sqsConsumer) Handler(baseCtx context.Context, lambdaEvent events.SQSEvent) error {
//todo: create context with logger
rand.Seed(time.Now().Unix())
@@ -115,17 +112,18 @@ func (consumer consumer) Handler(baseCtx context.Context, lambdaEvent events.SQS
// defer sqs.crashReporter.Catch(ctx)
// }
- if consumer.ConstantMessageType != "" {
+ if c.ConstantMessageType != "" {
//legacy mode for fixed message type as used in shiplogic
//where the whole instance is started for a specific SQS_MESSAGE_TYPE defined in environment
- handler, err := consumer.router.Route(consumer.ConstantMessageType)
+ handler, err := c.router.Route(c.ConstantMessageType)
if err != nil {
- return errors.Wrapf(err, "messageType=%s not handled", consumer.ConstantMessageType) //checked on startup - should never get here!!!
+ return errors.Wrapf(err, "messageType=%s not handled", c.ConstantMessageType) //checked on startup - should never get here!!!
}
if msgHandler, ok := handler.(func(events.SQSEvent) error); !ok {
- return errors.Wrapf(err, "SQS_MESSAGE_TYPE=%s: handler signature %T not supported", consumer.ConstantMessageType, handler)
+ return errors.Wrapf(err, "SQS_MESSAGE_TYPE=%s: handler signature %T not supported", c.ConstantMessageType, handler)
} else {
+ //call the handler
return msgHandler(lambdaEvent)
}
} else {
@@ -133,20 +131,36 @@ func (consumer consumer) Handler(baseCtx context.Context, lambdaEvent events.SQS
//process all message records in this event:
for messageIndex, message := range lambdaEvent.Records {
//get request-id for this message record
+ startTime := time.Now()
requestID := ""
- if requestIDAttr, ok := message.MessageAttributes[consumer.requestIDHeaderKey]; ok {
+ if requestIDAttr, ok := message.MessageAttributes[c.requestIDHeaderKey]; ok {
requestID = *requestIDAttr.StringValue
}
messageType := ""
+ var requestToLog interface{}
+ var handlerErr error
+ requestToLog = message.Body //will be logged as string if failed before parsing body into struct
+ defer func() {
+ if err := logs.LogSQSRequest(
+ startTime,
+ requestID,
+ messageType,
+ requestToLog,
+ handlerErr,
+ ); err != nil {
+ c.Errorf("failed to log: %+v", err)
+ }
+ }()
+
if messageTypeAttr, ok := message.MessageAttributes["type"]; !ok || messageTypeAttr.StringValue == nil {
- consumer.Errorf("ignoring message without messageType") //todo: could support generic handler for these... not yet required
+ c.Errorf("ignoring message without messageType") //todo: could support generic handler for these... not yet required
continue
} else {
messageType = *messageTypeAttr.StringValue
}
- event := service.Event{
+ event := queues.Event{
//producer: nil,
MessageID: message.MessageId,
QueueName: "N/A", //not sure how to get queue name from lambda Event... would be good to log it, may be in os.Getenv(???)?
@@ -156,7 +170,7 @@ func (consumer consumer) Handler(baseCtx context.Context, lambdaEvent events.SQS
BodyJSON: message.Body,
}
- ctx, err := queues.NewContext(consumer.Service, event)
+ ctx, err := consumer.NewContext(c.Service, event)
if err != nil {
return err
}
@@ -167,12 +181,12 @@ func (consumer consumer) Handler(baseCtx context.Context, lambdaEvent events.SQS
}).Infof("Queue(%s) Start SQS Handler Event: %v", ctx.Event().QueueName, ctx.Event())
//routing on messageType
- sqsHandler, err := consumer.router.Route(messageType)
+ sqsHandler, err := c.router.Route(messageType)
if err != nil {
ctx.Errorf("Unhandled sqs messageType(%v): %v", messageType, err)
continue
}
- handler, ok := sqsHandler.(queues.Handler)
+ handler, ok := sqsHandler.(consumer.Handler)
if !ok {
ctx.Errorf("messageType(%v) unsupported signature: %T", messageType, sqsHandler)
continue
@@ -189,13 +203,15 @@ func (consumer consumer) Handler(baseCtx context.Context, lambdaEvent events.SQS
ctx.Errorf("invalid message: %+v", err)
continue
}
+ requestToLog = recordStruct //replace string log with structured log
ctx.Debugf("message (%T) %+v", recordStruct, recordStruct)
args = append(args, reflect.ValueOf(recordStruct))
results := handler.FuncValue.Call(args)
if len(results) > 0 && !results[0].IsNil() {
- ctx.Errorf("handler failed: %+v", results[0].Interface().(error))
+ handlerErr = results[0].Interface().(error)
+ ctx.Errorf("handler failed: %+v", handlerErr)
}
}
}
diff --git a/cron/cron.go b/cron/cron.go
index 5a9dff2..114e106 100644
--- a/cron/cron.go
+++ b/cron/cron.go
@@ -8,7 +8,6 @@ import (
"github.com/aws/aws-lambda-go/lambda"
"github.com/aws/aws-lambda-go/lambdacontext"
"github.com/google/uuid"
- "gitlab.com/uafrica/go-utils/audit"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/logger"
"gitlab.com/uafrica/go-utils/service"
@@ -44,17 +43,11 @@ type Cron struct {
}
//wrap Service.WithStarter to return cron, else cannot be chained
-func (cron Cron) WithStarter(name string, starter service.IStarter) Cron {
+func (cron Cron) WithStarter(name string, starter service.Starter) Cron {
cron.Service = cron.Service.WithStarter(name, starter)
return cron
}
-//wrap else cannot be chained
-func (cron Cron) WithAuditor(auditor audit.Auditor) Cron {
- cron.Service = cron.Service.WithAuditor(auditor)
- return cron
-}
-
//add a check to startup of each context
//they will be called in the sequence they were added
//if check return error, processing stops and err is returned
diff --git a/cron/handler.go b/cron/handler.go
index 956e14c..814a41e 100644
--- a/cron/handler.go
+++ b/cron/handler.go
@@ -22,7 +22,7 @@ func NewHandler(fnc interface{}) (Handler, error) {
return h, errors.Errorf("returns %d results instead of (error)", fncType.NumOut())
}
- //arg[0] must implement interface queues.Context
+ //arg[0] must implement interface consumer.Context
if fncType.In(0) != contextInterfaceType &&
!fncType.In(0).Implements(contextInterfaceType) {
return h, errors.Errorf("first arg %v does not implement %v", fncType.In(0), contextInterfaceType)
diff --git a/examples/core/api/main.go b/examples/core/api/main.go
index b690732..5fe8aa1 100644
--- a/examples/core/api/main.go
+++ b/examples/core/api/main.go
@@ -6,7 +6,6 @@ import (
"os"
"gitlab.com/uafrica/go-utils/api"
- "gitlab.com/uafrica/go-utils/audit"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/examples/core/app"
"gitlab.com/uafrica/go-utils/examples/core/db"
@@ -22,7 +21,6 @@ func main() {
WithCheck("maintenance", maint{}).
WithCheck("rate", rateLimiter{}).
WithCORS(cors{}).
- WithAuditor(audit.File(os.Stdout)).
WithEvents(app.QueueRoutes()). //only used when LOG_LEVEL="debug"
Run()
}
diff --git a/examples/core/app/users/users.go b/examples/core/app/users/users.go
index 93e094a..5599be9 100644
--- a/examples/core/app/users/users.go
+++ b/examples/core/app/users/users.go
@@ -11,7 +11,7 @@ import (
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/examples/core/email"
"gitlab.com/uafrica/go-utils/logger"
- "gitlab.com/uafrica/go-utils/service"
+ "gitlab.com/uafrica/go-utils/queues"
)
type User struct {
@@ -113,7 +113,7 @@ func Add(ctx api.Context, params noParams, newUser POSTUser) (User, error) {
Subject: "Welcome User",
Body: "Your account has been created",
}
- /*eventID*/ _, err := service.NewEvent(ctx, "notify").RequestID(ctx.RequestID()).Type("email").Delay(time.Second * 5).Params(map[string]string{}).Send(email)
+ /*eventID*/ _, err := queues.NewEvent(ctx, "notify").RequestID(ctx.RequestID()).Type("email").Delay(time.Second * 5).Params(map[string]string{}).Send(email)
if err != nil {
ctx.Errorf("failed to notify: %+v", err)
}
diff --git a/examples/core/cron/main.go b/examples/core/cron/main.go
index f4f400f..66a494b 100644
--- a/examples/core/cron/main.go
+++ b/examples/core/cron/main.go
@@ -25,7 +25,6 @@ func main() {
cron.New(app.CronRoutes()).
WithStarter("db", db.Connector("core")).
- //WithAuditor(audit{}).
Run(invokeArnPtr)
}
diff --git a/examples/core/db/database.go b/examples/core/db/database.go
index 8021c42..f694c23 100644
--- a/examples/core/db/database.go
+++ b/examples/core/db/database.go
@@ -7,14 +7,14 @@ import (
"gitlab.com/uafrica/go-utils/service"
)
-func Connector(dbName string) service.IStarter {
+func Connector(dbName string) service.Starter {
return &connector{
name: dbName,
conn: 0,
}
}
-//connector implements service.IStarter
+//connector implements service.Starter
type connector struct {
name string
conn int
diff --git a/examples/core/email/notify.go b/examples/core/email/notify.go
index 206f2be..07a1149 100644
--- a/examples/core/email/notify.go
+++ b/examples/core/email/notify.go
@@ -1,8 +1,6 @@
package email
-import (
- "gitlab.com/uafrica/go-utils/queues"
-)
+import "gitlab.com/uafrica/go-utils/service"
type Message struct {
From string
@@ -13,7 +11,7 @@ type Message struct {
Body string
}
-func Notify(ctx queues.Context, msg Message) error {
+func Notify(ctx service.Context, msg Message) error {
ctx.Debugf("Pretending to send email: %+v", msg)
return nil
}
diff --git a/examples/core/sqs/main.go b/examples/core/sqs/main.go
index 7a019a7..276fb51 100644
--- a/examples/core/sqs/main.go
+++ b/examples/core/sqs/main.go
@@ -4,10 +4,10 @@ import (
"flag"
"gitlab.com/uafrica/go-utils/config"
+ "gitlab.com/uafrica/go-utils/consumer/sqs_consumer"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/examples/core/db"
"gitlab.com/uafrica/go-utils/logger"
- "gitlab.com/uafrica/go-utils/queues/sqs"
)
func main() {
@@ -16,7 +16,7 @@ func main() {
sqsRoutes := map[string]interface{}{}
- consumer := sqs.NewConsumer("uafrica-request-id", sqsRoutes).
+ consumer := sqs_consumer.New("uafrica-request-id", sqsRoutes).
WithStarter("db", db.Connector("core"))
if reqFile != nil && *reqFile != "" {
diff --git a/logs/README.md b/logs/README.md
new file mode 100644
index 0000000..75ec012
--- /dev/null
+++ b/logs/README.md
@@ -0,0 +1,43 @@
+# Logs
+
+This package provides functions to log API, SQS and CRON events, capturing for example an API method and path along with request and response details, or SQS event details along with the time it spent in the queue before being processed.
+
+These logs are sent to a SQS queue. The handling of the queue event and capturing of the log, is done as SQS handlers, not as part of this package.
+
+Do not confuse this with logger, which write free-format log entries to stdout.
+
+# API-LOGS
+
+At the end of each API handler, an api-log is captured to describe the incoming API request and response. Also part of this log is a list of actions takens during the handler, including:
+* API calls made
+* SQS Events sent
+* Database Statements executed
+* Time slept
+
+Each of those has a start/end time and duration, and they are summed and it should add up to most of the API-Log total duration.
+If there is a big difference between the summed time and the total duration, then we are doing something that takes time that we are not monitoring which should be investigated.
+The total time spent sleeping, waiting for db calls, waiting for outgoing API calls, is logged in the API log.
+This can be logged and broken down per path and method to see where the API is spending most time, and that could be investigated and optimised to improvie performance.
+
+# SQS-LOGS
+
+SQS logs are written at the end of a SQS event handler, similar to API logs.
+Since SQS is used to write API logs, those handlers should not be logged, as it will create a circular infinite queue ramping up huge bills.
+To be safe, SQS logs are therefore DISABLED by default.
+It should only be enabled for things like provider rate requests or any SQS handler that is part of the functionality of the system doing async work, not handlers that are part of the infrastructure.
+
+SQS log will also write to the API_LOGS queue and the same index in OpenSearch (can review and change this in future)
+It logs with method "SQS" and path is the message type.
+That means we can log durations and through put in the same way and on the same graph as API when needed
+It also captures the actions taken as part of the handler, in the same way it is captured for API.
+
+So when one finds some action takes too long in API, and move it to an SQS handler, the change will be visibile on the dashboard and indicate the improvement or not if your change did not have the desired effect.
+
+That it is idea.
+
+We can easily disable SQS logs and we can easily move it to another index in OpenSearch if necessary. Will have to try it for a while an see if it is useful in the current form or not.
+
+# CRON-LOGS
+
+In the same way we log API/SQS, it will be useful to monitor crons with a bit of output, e.g. nr of items deleted by a janitor etc.
+One can get that currently from cloud watch if the logs are not disabled, and CloudWatch should ideally not be source of metrics, but that is currently the case, so not changing it yet.
\ No newline at end of file
diff --git a/logs/api-logs.go b/logs/api-logs.go
new file mode 100644
index 0000000..a2e6d47
--- /dev/null
+++ b/logs/api-logs.go
@@ -0,0 +1,230 @@
+package logs
+
+import (
+ "net/http"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/aws/aws-lambda-go/events"
+ "gitlab.com/uafrica/go-utils/errors"
+ "gitlab.com/uafrica/go-utils/queues"
+ "gitlab.com/uafrica/go-utils/service"
+)
+
+//Call this at the end of an API request handler to capture the req/res as well as all actions taken during the processing
+//(note: action list is only reset when this is called - so must be called after each handler, else action list has to be reset at the start)
+func LogIncomingAPIRequest(startTime time.Time, req events.APIGatewayProxyRequest, res events.APIGatewayProxyResponse) error {
+ if service.Ctx == nil {
+ return errors.Errorf("Cannot log without service context")
+ }
+
+ //todo: filter out some noisy (method+path)
+
+ endTime := time.Now()
+ dur := endTime.Sub(startTime)
+
+ var authType string
+ var authUsername string
+ if req.RequestContext.Identity.CognitoAuthenticationType != "" {
+ authType = "cognito"
+ split := strings.Split(req.RequestContext.Identity.CognitoAuthenticationProvider, ":")
+ if len(split) > 0 {
+ authUsername = split[len(split)-1] //= part after last ':'
+ }
+ } else {
+ authType = "iam"
+ split := strings.Split(req.RequestContext.Identity.UserArn, ":user/")
+ if len(split) > 0 {
+ authUsername = split[len(split)-1] //= part after ':user/'
+ }
+ }
+
+ claim := service.Ctx.Claim()
+ username, _ := claim["user_id"].(string)
+ accountID, _ := claim["account_id"].(int64)
+ apiLog := ApiLog{
+ StartTime: startTime,
+ EndTime: endTime,
+ DurMs: int(dur / time.Millisecond),
+ Method: req.HTTPMethod,
+ Address: req.RequestContext.DomainName,
+ Path: req.Path,
+ ResponseCode: res.StatusCode,
+ RequestID: service.Ctx.RequestID(),
+ InitialAuthType: authType,
+ InitialAuthUsername: authUsername,
+ SourceIP: req.RequestContext.Identity.SourceIP,
+ UserAgent: req.RequestContext.Identity.UserAgent,
+ Username: username,
+ AccountID: accountID,
+ Request: ApiLogRequest{
+ Headers: req.Headers,
+ QueryParameters: req.QueryStringParameters,
+ BodySize: len(req.Body),
+ Body: req.Body,
+ },
+ Response: ApiLogResponse{
+ Headers: res.Headers,
+ BodySize: len(res.Body),
+ Body: res.Body,
+ },
+ //RelevantID: service.Ctx.,???
+ }
+
+ //copy then reset actionList for the next handler
+ actionListMutex.Lock()
+ apiLog.Actions = actionList
+ actionList = []ActionLog{}
+ actionListMutex.Unlock()
+
+ //also copy multi-value query parameters to the log as CSV array values
+ for n, as := range req.MultiValueQueryStringParameters {
+ apiLog.Request.QueryParameters[n] = "[" + strings.Join(as, ",") + "]"
+ }
+
+ //todo: filter out excessive req/res body content per (method+path)
+ //todo: also need to do for all actions...
+ if apiLog.Method == http.MethodGet {
+ apiLog.Response.Body = "<not logged>"
+ }
+
+ //todo: filter out sensitive values (e.g. OTP)
+ if _, err := queues.NewEvent(service.Ctx, "API_LOGS").
+ Type("incoming-api-log").
+ RequestID(apiLog.RequestID).
+ Send(apiLog); err != nil {
+ return errors.Wrapf(err, "failed to send incoming-api-log")
+ }
+ return nil
+} //LogIncomingAPIRequest()
+
+//ApiLog is the SQS event details struct encoded as JSON document, sent to SQS, to be logged for each API handler executed.
+type ApiLog struct {
+ StartTime time.Time `json:"start_time"`
+ EndTime time.Time `json:"end_time"`
+ DurMs int `json:"duration_ms"` //duration in milliseconds
+ Method string `json:"method"`
+ Address string `json:"address"` //server address for incoming and outgoing
+ Path string `json:"path"`
+ ResponseCode int `json:"response_code"`
+ RequestID string `json:"request_id"`
+ InitialAuthUsername string `json:"initial_auth_username,omitempty"`
+ InitialAuthType string `json:"initial_auth_type,omitempty"`
+ AccountID int64 `json:"account_id,omitempty"`
+ Username string `json:"username,omitempty"`
+ SourceIP string `json:"source_ip,omitempty"` //only logged for incoming API
+ UserAgent string `json:"user_agent,omitempty"` //only for incoming, indicate type of browser when UI
+ RelevantID string `json:"relevant_id,omitempty"`
+ Request ApiLogRequest `json:"request"`
+ Response ApiLogResponse `json:"response"`
+ Actions []ActionLog `json:"actions,omitempty"`
+}
+
+type ApiLogRequest struct {
+ Headers map[string]string `json:"headers,omitempty"`
+ QueryParameters map[string]string `json:"query_parameters,omitempty"`
+ BodySize int `json:"body_size"` //set even when body is truncated/omitted
+ Body string `json:"body,omitempty"` //json body as a string
+}
+
+type ApiLogResponse struct {
+ Headers map[string]string `json:"headers,omitempty"`
+ BodySize int `json:"body_size"` //set even when body is truncated/omitted
+ Body string `json:"body,omitempty"` //json content as a string
+}
+
+func LogOutgoingAPIRequest(url string, method string, requestBody string, responseBody string, responseCode int, startTime time.Time) error {
+ endTime := time.Now()
+ dur := endTime.Sub(startTime)
+
+ apiCallLog := ApiCallLog{
+ URL: url,
+ Method: method,
+ ResponseCode: responseCode,
+ Request: ApiCallRequestLog{
+ BodySize: len(requestBody),
+ Body: requestBody,
+ },
+ Response: ApiCallResponseLog{
+ BodySize: len(responseBody),
+ Body: responseBody,
+ },
+ }
+
+ actionListMutex.Lock()
+ actionList = append(actionList, ActionLog{
+ StartTime: startTime,
+ EndTime: endTime,
+ DurMs: int(dur / time.Millisecond),
+ Type: ActionTypeApiCall,
+ ApiCall: &apiCallLog,
+ })
+ actionListMutex.Unlock()
+
+ return nil
+} //LogOutgoingAPIRequest()
+
+var (
+ actionListMutex sync.Mutex
+ actionList = []ActionLog{}
+)
+
+type ActionLog struct {
+ StartTime time.Time `json:"start_time"`
+ EndTime time.Time `json:"end_time"`
+ DurMs int `json:"duration_ms"` //duration in milliseconds
+ Type ActionType `json:"type" doc:"Type is api-call|sqs-sent|sql-query|sleep"`
+ ApiCall *ApiCallLog `json:"api_call,omitempty"`
+ SQSSent *SQSSentLog `json:"sqs_sent,omitempty"`
+ SQLQuery *SQLQueryLog `json:"sql_query,omitempty"`
+ Sleep *SleepLog `json:"sleep,omitempty"`
+}
+
+type ActionType string
+
+var ActionTypeList = []ActionType{
+ ActionTypeApiCall,
+ ActionTypeSqsSent,
+ ActionTypeSqlQuery,
+ ActionTypeSleep,
+}
+
+const (
+ ActionTypeApiCall ActionType = "api-call"
+ ActionTypeSqsSent ActionType = "sqs-sent"
+ ActionTypeSqlQuery ActionType = "sql-query"
+ ActionTypeSleep ActionType = "sleep"
+)
+
+//APICallLog captures details of an outgoing API call made from a handler
+type ApiCallLog struct {
+ URL string `json:"url"`
+ Method string `json:"method"`
+ ResponseCode int `json:"response_code"`
+ Request ApiCallRequestLog `json:"request"`
+ Response ApiCallResponseLog `json:"response"`
+}
+
+type ApiCallRequestLog struct {
+ BodySize int `json:"body_size"`
+ Body string `json:"body"`
+}
+
+type ApiCallResponseLog struct {
+ BodySize int `json:"body_size"`
+ Body string `json:"body"`
+}
+
+//SQSSentLog captures details of an SQS event sent from a handler
+type SQSSentLog struct {
+}
+
+//SQLQueryLog captures details of an SQL query executed from a handler
+type SQLQueryLog struct {
+}
+
+//SleepLog captures details of time spent sleeping from a handler
+type SleepLog struct {
+ //nothing to record apart from the action start/end/dur...
+}
diff --git a/logs/cron.logs.go b/logs/cron.logs.go
new file mode 100644
index 0000000..1f58fd5
--- /dev/null
+++ b/logs/cron.logs.go
@@ -0,0 +1,3 @@
+package logs
+
+//todo... currently monitored from CloudWatch...
diff --git a/logs/sqs-logs.go b/logs/sqs-logs.go
new file mode 100644
index 0000000..49f446b
--- /dev/null
+++ b/logs/sqs-logs.go
@@ -0,0 +1,96 @@
+package logs
+
+import (
+ "encoding/json"
+ "fmt"
+ "os"
+ "time"
+
+ "gitlab.com/uafrica/go-utils/errors"
+ "gitlab.com/uafrica/go-utils/logger"
+ "gitlab.com/uafrica/go-utils/queues"
+ "gitlab.com/uafrica/go-utils/service"
+)
+
+//Call this at the end of an SQS event handler to capture the req and result as well as all actions taken during the processing
+//(note: action list is only reset when this is called - so must be called after each handler, else action list has to be reset at the start)
+func LogSQSRequest(startTime time.Time,
+ requestID string, //from API
+ messageType string,
+ req interface{},
+ handlerErr error,
+) error {
+ if service.Ctx == nil {
+ return errors.Errorf("Cannot log without service context")
+ }
+
+ if !sqsLogEnabled {
+ return nil
+ }
+
+ endTime := time.Now()
+ dur := endTime.Sub(startTime)
+ log := ApiLog{
+ StartTime: startTime,
+ EndTime: endTime,
+ DurMs: int(dur / time.Millisecond),
+ RequestID: requestID,
+ Method: "SQS",
+ Path: messageType,
+ }
+
+ if req != nil {
+ if jsonReq, err := json.Marshal(req); err == nil {
+ log.Request.Body = string(jsonReq)
+ log.Request.BodySize = len(log.Request.Body)
+ }
+ }
+
+ if handlerErr == nil {
+ log.ResponseCode = 0
+ } else {
+ log.ResponseCode = 1
+ errorInfo := ErrorInfo{
+ Error: handlerErr.Error(),
+ Details: fmt.Sprintf("%+v", handlerErr),
+ }
+ jsonError, _ := json.Marshal(errorInfo)
+ log.Response.Body = string(jsonError)
+ }
+
+ //copy then reset actionList for the next handler
+ actionListMutex.Lock()
+ log.Actions = actionList
+ actionList = []ActionLog{}
+ actionListMutex.Unlock()
+
+ //todo: filter out sensitive values (e.g. OTP)
+
+ //note: we send SQS logs to "API_LOGS" which already exists... should be renamed to simply "LOGS"
+ //it use the same structure, but method="SQS" and path="messageType" and request is the event body
+ //so they can be plotted on the same dashboard visualisation in OpenSearch with all the same filters/metrics
+ if _, err := queues.NewEvent(service.Ctx, "API_LOGS").
+ Type("api-log").
+ RequestID(requestID).
+ Send(log); err != nil {
+ return errors.Wrapf(err, "failed to send api-log for SQS")
+ }
+ return nil
+}
+
+var sqsLogEnabled = false
+
+func init() {
+ envSetting := os.Getenv("SQS_LOGS_ENABLED")
+ if envSetting == "true" {
+ sqsLogEnabled = true
+ }
+ //if consuming from API_LOGS, do not enable else we will consume and send to our own queue!
+
+ logger.Infof("Environment SQS_LOGS_ENABLED=\"%s\" -> sqsLogsEnabled=%v", envSetting, sqsLogEnabled)
+}
+
+type ErrorInfo struct {
+ Error string `json:"error"`
+ Details string `json:"details"`
+}
diff --git a/queues/audit.go b/queues/audit.go
deleted file mode 100644
index 16d5009..0000000
--- a/queues/audit.go
+++ /dev/null
@@ -1,55 +0,0 @@
-package queues
-
-import (
- "time"
-
- "gitlab.com/uafrica/go-utils/audit"
- "gitlab.com/uafrica/go-utils/errors"
- "gitlab.com/uafrica/go-utils/service"
-)
-
-//create auditor that push to a queue using the specified producer
-func Auditor(queueName string, messageType string, producer service.ProducerLogger) audit.Auditor {
- if producer == nil {
- panic(errors.Errorf("cannot create auditor with producer=nil"))
- }
- if queueName == "" {
- queueName = "AUDIT"
- }
- if messageType == "" {
- messageType = "audit"
- }
- return auditor{
- producer: producer,
- queueName: queueName,
- messageType: messageType,
- }
-}
-
-type auditor struct {
- producer service.ProducerLogger
- queueName string
- messageType string
-}
-
-func (a auditor) WriteEvent(requestID string, event audit.Event) error {
- _, err := service.NewEvent(a.producer, a.queueName).
- RequestID(requestID).
- Type(a.messageType).
- Send(event)
- if err != nil {
- return errors.Wrapf(err, "failed to write audit event")
- }
- return nil
-}
-
-func (a auditor) WriteValues(startTime, endTime time.Time, requestID string, values map[string]interface{}) error {
- _, err := service.NewEvent(a.producer, a.queueName).
- RequestID(requestID).
- Type(a.messageType).
- Send(values)
- if err != nil {
- return errors.Wrapf(err, "failed to write audit values")
- }
- return nil
-}
diff --git a/queues/check.go b/queues/check.go
deleted file mode 100644
index f0148ee..0000000
--- a/queues/check.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package queues
-
-type ICheck interface {
- Check(Context) (interface{}, error)
-}
diff --git a/service/event.go b/queues/event.go
similarity index 91%
rename from service/event.go
rename to queues/event.go
index 08f6c98..376947d 100644
--- a/service/event.go
+++ b/queues/event.go
@@ -1,4 +1,4 @@
-package service
+package queues
import (
"encoding/json"
@@ -9,12 +9,7 @@ import (
"gitlab.com/uafrica/go-utils/logger"
)
-type ProducerLogger interface {
- Producer
- logger.Logger
-}
-
-func NewEvent(producer ProducerLogger, queueName string) Event {
+func NewEvent(producer Producer, queueName string) Event {
if producer == nil {
panic(errors.Errorf("NewEvent(producer=nil)"))
}
@@ -31,7 +26,7 @@ func NewEvent(producer ProducerLogger, queueName string) Event {
}
type Event struct {
- producer ProducerLogger
+ producer Producer
MessageID string //assigned by implementation (AWS/mem/..)
QueueName string //queue determine sequencing, items in same queue are delivered one-after-the-other, other queues may deliver concurrent to this queue
TypeName string //type determines which handler processes the event
@@ -81,6 +76,8 @@ func (event Event) Params(params map[string]string) Event {
return event
}
+var log = logger.New()
+
func (event Event) Send(value interface{}) (string, error) {
if event.producer == nil {
return "", errors.Errorf("send with producer==nil")
@@ -95,7 +92,7 @@ func (event Event) Send(value interface{}) (string, error) {
if msgID, err := event.producer.Send(event); err != nil {
return "", errors.Wrapf(err, "failed to send event")
} else {
- event.producer.WithFields(map[string]interface{}{
+ log.WithFields(map[string]interface{}{
"queue": event.QueueName,
"type": event.TypeName,
"due": event.DueTime,
diff --git a/service/producer.go b/queues/producer.go
similarity index 55%
rename from service/producer.go
rename to queues/producer.go
index 050c56f..00df1db 100644
--- a/service/producer.go
+++ b/queues/producer.go
@@ -1,7 +1,7 @@
-package service
+package queues
//Producer sends an event for async processing
type Producer interface {
+ NewEvent(queueName string) Event
Send(event Event) (msgID string, err error)
- //todo: method to request an event after some delay with incrementing attempt nr
}
diff --git a/queues/sqs/README.md b/queues/sqs_producer/README.md
similarity index 100%
rename from queues/sqs/README.md
rename to queues/sqs_producer/README.md
diff --git a/queues/sqs/producer.go b/queues/sqs_producer/producer.go
similarity index 90%
rename from queues/sqs/producer.go
rename to queues/sqs_producer/producer.go
index ce10639..55a0fd7 100644
--- a/queues/sqs/producer.go
+++ b/queues/sqs_producer/producer.go
@@ -1,4 +1,4 @@
-package sqs
+package sqs_producer
import (
"os"
@@ -11,10 +11,10 @@ import (
"github.com/aws/aws-sdk-go/service/sqs"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/logger"
- "gitlab.com/uafrica/go-utils/service"
+ "gitlab.com/uafrica/go-utils/queues"
)
-func NewProducer(requestIDHeaderKey string) service.Producer {
+func New(requestIDHeaderKey string) queues.Producer {
region := os.Getenv("AWS_REGION")
if region == "" {
panic(errors.Errorf("environment AWS_REGION is not defined"))
@@ -39,8 +39,12 @@ type producer struct {
queues map[string]*QueueProducer
}
+func (producer *producer) NewEvent(queueName string) queues.Event {
+ return queues.NewEvent(producer, queueName)
+}
+
// Note: Calling code needs SQS IAM permissions
-func (producer *producer) Send(event service.Event) (string, error) {
+func (producer *producer) Send(event queues.Event) (string, error) {
logger.Debugf("SQS producer.Send(%+v)", event)
messenger, ok := producer.queues[event.QueueName]
if !ok {
@@ -89,7 +93,7 @@ type QueueProducer struct {
queueURL string
}
-func (m *QueueProducer) Send(event service.Event) (string, error) {
+func (m *QueueProducer) Send(event queues.Event) (string, error) {
logger.Debugf("SQS producer.queue(%s) Sending event %+v", m.queueURL, event)
//add params as message attributes
diff --git a/service/README.md b/service/README.md
index ce8c48c..60b637e 100644
--- a/service/README.md
+++ b/service/README.md
@@ -34,7 +34,7 @@ Example:
Where package db then defines:
- func Connector(dbName string) service.IStarter {
+ func Connector(dbName string) service.Starter {
return &connector{
dbName: dbName,
dbConn: nil,
@@ -157,16 +157,11 @@ So:
```claim,ok := ctx.Get("claims").(MyClaimStruct)```
* All fields in params or body structs matching claim names will be overwritten by the time a handler is called.
-# Audits
-
-Audit records are written with:
-* ctx.AuditChange(), or
-* ctx.AuditWrite()
+# Data Change Audits
+Data Change audit records are written with ctx.AuditChange()
The AuditChange() method logs the changes between an original and new value.
-The AuditWrite() logs all the data given to it.
-A handler may write 0..N audit record, there is no check. In general, audits are written to capture changes, and when a handler changes multiple database records, they could all be audited.
# Sending Async Events
diff --git a/service/context.go b/service/context.go
index ce4acc2..533af78 100644
--- a/service/context.go
+++ b/service/context.go
@@ -9,6 +9,7 @@ import (
"gitlab.com/uafrica/go-utils/audit"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/logger"
+ "gitlab.com/uafrica/go-utils/queues"
"gitlab.com/uafrica/go-utils/string_utils"
)
@@ -18,8 +19,7 @@ var Ctx Context
type Context interface {
context.Context
logger.Logger
- Producer
- audit.Auditor
+ queues.Producer
RequestID() string
MillisecondsSinceStart() int64
@@ -51,7 +51,7 @@ type Context interface {
ValueOrDefault(name string, defaultValue interface{}) interface{}
Data() map[string]interface{}
- //write an audit event
+ //write a data change audit event
AuditChange(eventType string, orgValue, newValue interface{})
}
@@ -78,7 +78,6 @@ func (s service) NewContext(base context.Context, requestID string, values map[s
Context: base,
Logger: l,
Producer: s.Producer,
- Auditor: s.Auditor,
startTime: time.Now(),
requestID: requestID,
data: map[string]interface{}{},
@@ -99,15 +98,13 @@ func (s service) NewContext(base context.Context, requestID string, values map[s
Ctx.Debugf("Start(%s)=(%T)%+v", starterName, starterData, starterData)
}
-
return Ctx, nil
}
type serviceContext struct {
context.Context
logger.Logger
- Producer
- audit.Auditor
+ queues.Producer
startTime time.Time
requestID string
claim map[string]interface{}
@@ -220,17 +217,14 @@ func (ctx *serviceContext) ValueOrDefault(name string, defaultValue interface{})
func (ctx *serviceContext) AuditChange(eventType string, orgValue, newValue interface{}) {
username, _ := ctx.Claim()["username"].(string)
- event, err := audit.NewEvent(
+ if err := audit.SaveDataChange(
+ ctx.requestID,
username, //use username as source (will default to "SYSTEM" if undefined)
eventType,
orgValue,
newValue,
- )
- if err != nil {
- ctx.Errorf("failed to define audit event: %+v", err)
+ ); err != nil {
+ ctx.Errorf("failed to save data change: %+v", err)
return
}
- if err := ctx.Auditor.WriteEvent(ctx.requestID, event); err != nil {
- ctx.Errorf("failed to audit change: %+v", err)
- }
}
diff --git a/service/service.go b/service/service.go
index 0a80933..3eda7b0 100644
--- a/service/service.go
+++ b/service/service.go
@@ -4,19 +4,17 @@ import (
"context"
"os"
- "gitlab.com/uafrica/go-utils/audit"
"gitlab.com/uafrica/go-utils/errors"
"gitlab.com/uafrica/go-utils/logger"
+ "gitlab.com/uafrica/go-utils/queues"
"gitlab.com/uafrica/go-utils/string_utils"
)
type Service interface {
logger.Logger
- Producer
- audit.Auditor
- WithStarter(name string, starter IStarter) Service
- WithProducer(producer Producer) Service
- WithAuditor(auditor audit.Auditor) Service
+ queues.Producer
+ WithStarter(name string, starter Starter) Service
+ WithProducer(producer queues.Producer) Service
NewContext(base context.Context, requestID string, values map[string]interface{}) (Context, error)
}
@@ -28,18 +26,16 @@ func New() Service {
return service{
Producer: nil,
Logger: logger.New().WithFields(map[string]interface{}{"env": env}),
- Auditor: audit.None(),
env: env,
- starters: map[string]IStarter{},
+ starters: map[string]Starter{},
}
}
type service struct {
- logger.Logger //for logging outside of context
- Producer //for sending async events
- audit.Auditor
- env string
- starters map[string]IStarter
+ logger.Logger //for logging outside of context
+ queues.Producer //for sending async events
+ env string
+ starters map[string]Starter
}
func (s service) Env() string {
@@ -59,7 +55,7 @@ func (s service) Env() string {
//you can implement one starter that does everything and return a struct or
//implement one for your db, one for rate limit, one for ...
//the name must be snake-case, e.g. "this_is_my_starter_name"
-func (s service) WithStarter(name string, starter IStarter) Service {
+func (s service) WithStarter(name string, starter Starter) Service {
if !string_utils.IsSnakeCase(name) {
panic(errors.Errorf("invalid starter name=\"%s\", expecting snake_case names only", name))
}
@@ -73,16 +69,9 @@ func (s service) WithStarter(name string, starter IStarter) Service {
return s
}
-func (s service) WithProducer(producer Producer) Service {
+func (s service) WithProducer(producer queues.Producer) Service {
if producer != nil {
s.Producer = producer
}
return s
}
-
-func (s service) WithAuditor(auditor audit.Auditor) Service {
- if auditor != nil {
- s.Auditor = auditor
- }
- return s
-}
diff --git a/service/start.go b/service/start.go
index c30c117..143a4e1 100644
--- a/service/start.go
+++ b/service/start.go
@@ -1,6 +1,6 @@
package service
-type IStarter interface {
+type Starter interface {
//called at the start of api/cron/queues processing, before checks, e.g. to ensure we have db connection
//i.e. setup things that does not depend on the request/event details
//if you need the request details, you need to implement a check for each of the api, cron and/or queue as needed, not a Start() method.
--
GitLab