Context package. Monitor

Author: aopoltorzhicky

config/config.go

@@ -15,7 +15,8 @@ type Config struct {
 	Database    Database              `yaml:"database" validate:"required"`
 	DataSources map[string]DataSource `yaml:"datasources"`
 	Contracts   map[string]Contract   `yaml:"contracts"`
-	Hasura      Hasura                `yaml:"hasura"`
+	Hasura      Hasura                `yaml:"hasura" validate:"omitempty"`
+	Prometheus  Prometheus            `yaml:"prometheus" validate:"omitempty"`
 }
 
 // Substitute -
@@ -56,6 +57,11 @@ type Hasura struct {
 	Rest               *bool  `yaml:"rest"`
 }
 
+// Prometheus -
+type Prometheus struct {
+	URL string `yaml:"url" validate:"required"`
+}
+
 // Load - load default config from `filename`
 func Load(filename string) (*Config, error) {
 	if filename == "" {

go.sum

@@ -112,11 +112,13 @@ github.com/lib/pq v1.3.0 h1:/qkRGz8zljWiDcFvgpwUpwIAPu3r07TDvs3Rws+o/pU=
 github.com/lib/pq v1.3.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/mattn/go-colorable v0.1.1/go.mod h1:FuOcm+DKB9mbwrcAfNl7/TZVBZ6rcnceauSikq3lYCQ=
 github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
+github.com/mattn/go-colorable v0.1.6 h1:6Su7aK7lXmJ/U79bYtBjLNaha4Fs1Rg9plHpcH+vvnE=
 github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-isatty v0.0.5/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
 github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
+github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
 github.com/mattn/go-sqlite3 v1.14.5 h1:1IdxlwTNazvbKJQSxoJ5/9ECbEeaTTyeU7sEAZ5KKTQ=
 github.com/mattn/go-sqlite3 v1.14.5/go.mod h1:WVKg1VTActs4Qso6iwGbiFih2UIHo0ENGwNd0Lj+XmI=

hasura/api.go

@@ -2,6 +2,7 @@ package hasura
 
 import (
 	"bytes"
+	"context"
 	"net/http"
 	"net/url"
 	"path"
@@ -41,12 +42,12 @@ func (api *API) buildURL(endpoint string, args map[string]string) (string, error
 	return u.String(), nil
 }
 
-func (api *API) get(endpoint string, args map[string]string) (*http.Response, error) {
+func (api *API) get(ctx context.Context, endpoint string, args map[string]string) (*http.Response, error) {
 	url, err := api.buildURL(endpoint, args)
 	if err != nil {
 		return nil, err
 	}
-	req, err := http.NewRequest(http.MethodGet, url, nil)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -55,7 +56,7 @@ func (api *API) get(endpoint string, args map[string]string) (*http.Response, er
 }
 
 //nolint
-func (api *API) post(endpoint string, args map[string]string, body interface{}, output interface{}) error {
+func (api *API) post(ctx context.Context, endpoint string, args map[string]string, body interface{}, output interface{}) error {
 	url, err := api.buildURL(endpoint, args)
 	if err != nil {
 		return err
@@ -66,7 +67,7 @@ func (api *API) post(endpoint string, args map[string]string, body interface{},
 		return err
 	}
 
-	req, err := http.NewRequest(http.MethodPost, url, bytes.NewBuffer(postBody))
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewBuffer(postBody))
 	if err != nil {
 		return err
 	}
@@ -97,8 +98,8 @@ func (api *API) post(endpoint string, args map[string]string, body interface{},
 }
 
 // Health
-func (api *API) Health() error {
-	resp, err := api.get("/healthz", nil)
+func (api *API) Health(ctx context.Context) error {
+	resp, err := api.get(ctx, "/healthz", nil)
 	if err != nil {
 		return err
 	}
@@ -111,24 +112,24 @@ func (api *API) Health() error {
 }
 
 // ExportMetadata -
-func (api *API) ExportMetadata(data *Metadata) (ExportMetadataResponse, error) {
+func (api *API) ExportMetadata(ctx context.Context, data *Metadata) (ExportMetadataResponse, error) {
 	req := request{
 		Type: "export_metadata",
 		Args: data,
 	}
 	var resp ExportMetadataResponse
-	err := api.post("/v1/query", nil, req, &resp)
+	err := api.post(ctx, "/v1/query", nil, req, &resp)
 	return resp, err
 }
 
 // ReplaceMetadata -
-func (api *API) ReplaceMetadata(data *Metadata) error {
+func (api *API) ReplaceMetadata(ctx context.Context, data *Metadata) error {
 	req := request{
 		Type: "replace_metadata",
 		Args: data,
 	}
 	var resp replaceMetadataResponse
-	if err := api.post("/v1/query", nil, req, &resp); err != nil {
+	if err := api.post(ctx, "/v1/query", nil, req, &resp); err != nil {
 		return err
 	}
 	if resp.Message == "success" {
@@ -138,19 +139,19 @@ func (api *API) ReplaceMetadata(data *Metadata) error {
 }
 
 // TrackTable -
-func (api *API) TrackTable(schema, name string) error {
+func (api *API) TrackTable(ctx context.Context, schema, name string) error {
 	req := request{
 		Type: "track_table",
 		Args: map[string]string{
 			"schema": schema,
 			"name":   name,
 		},
 	}
-	return api.post("/v1/query", nil, req, nil)
+	return api.post(ctx, "/v1/query", nil, req, nil)
 }
 
 // CreateSelectPermissions - A select permission is used to restrict access to only the specified columns and rows.
-func (api *API) CreateSelectPermissions(table, role string, perm Permission) error {
+func (api *API) CreateSelectPermissions(ctx context.Context, table, role string, perm Permission) error {
 	req := request{
 		Type: "create_select_permission",
 		Args: map[string]interface{}{
@@ -159,23 +160,23 @@ func (api *API) CreateSelectPermissions(table, role string, perm Permission) err
 			"permission": perm,
 		},
 	}
-	return api.post("/v1/query", nil, req, nil)
+	return api.post(ctx, "/v1/query", nil, req, nil)
 }
 
 // DropSelectPermissions -
-func (api *API) DropSelectPermissions(table, role string) error {
+func (api *API) DropSelectPermissions(ctx context.Context, table, role string) error {
 	req := request{
 		Type: "drop_select_permission",
 		Args: map[string]interface{}{
 			"table": table,
 			"role":  role,
 		},
 	}
-	return api.post("/v1/query", nil, req, nil)
+	return api.post(ctx, "/v1/query", nil, req, nil)
 }
 
 // CreateRestEndpoint -
-func (api *API) CreateRestEndpoint(name, url, queryName, collectionName string) error {
+func (api *API) CreateRestEndpoint(ctx context.Context, name, url, queryName, collectionName string) error {
 	req := request{
 		Type: "create_rest_endpoint",
 		Args: map[string]interface{}{
@@ -190,5 +191,5 @@ func (api *API) CreateRestEndpoint(name, url, queryName, collectionName string)
 			},
 		},
 	}
-	return api.post("/v1/query", nil, req, nil)
+	return api.post(ctx, "/v1/query", nil, req, nil)
 }

hasura/hasura.go

@@ -1,6 +1,7 @@
 package hasura
 
 import (
+	"context"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -19,23 +20,39 @@ const (
 	allowedQueries = "allowed-queries"
 )
 
+func checkHealth(ctx context.Context, api *API) {
+	log.Info("Waiting hasura is up and running")
+	if err := api.Health(ctx); err != nil {
+		return
+	}
+	ticker := time.NewTicker(5 * time.Second)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := api.Health(ctx); err != nil {
+				log.Warn(err)
+				continue
+			}
+			return
+		}
+	}
+}
+
 // Create - creates hasura models
-func Create(hasura config.Hasura, cfg config.Database, views []string, models ...interface{}) error {
+func Create(ctx context.Context, hasura config.Hasura, cfg config.Database, views []string, models ...interface{}) error {
 	api := New(hasura.URL, hasura.Secret)
 
-	log.Info("Waiting hasura is up and running")
-	for err := api.Health(); err != nil; err = api.Health() {
-		log.Warn(err)
-		time.Sleep(time.Second * 5)
-	}
+	checkHealth(ctx, api)
 
 	metadata, err := Generate(hasura, cfg, models...)
 	if err != nil {
 		return err
 	}
 
 	log.Info("Fetching existing metadata...")
-	export, err := api.ExportMetadata(metadata)
+	export, err := api.ExportMetadata(ctx, metadata)
 	if err != nil {
 		return err
 	}
@@ -57,14 +74,14 @@ func Create(hasura config.Hasura, cfg config.Database, views []string, models ..
 	}
 
 	log.Info("Replacing metadata...")
-	if err := api.ReplaceMetadata(metadata); err != nil {
+	if err := api.ReplaceMetadata(ctx, metadata); err != nil {
 		return err
 	}
 
 	if len(metadata.QueryCollections) > 0 && (hasura.Rest == nil || *hasura.Rest) {
 		log.Info("Creating REST endpoints...")
 		for _, query := range metadata.QueryCollections[0].Definition.Queries {
-			if err := api.CreateRestEndpoint(query.Name, query.Name, query.Name, allowedQueries); err != nil {
+			if err := api.CreateRestEndpoint(ctx, query.Name, query.Name, query.Name, allowedQueries); err != nil {
 				if e, ok := err.(APIError); !ok || !e.AlreadyExists() {
 					return err
 				}
@@ -74,15 +91,15 @@ func Create(hasura config.Hasura, cfg config.Database, views []string, models ..
 
 	log.Info("Tracking views...")
 	for i := range views {
-		if err := api.TrackTable("public", views[i]); err != nil {
+		if err := api.TrackTable(ctx, "public", views[i]); err != nil {
 			if !strings.Contains(err.Error(), "view/table already tracked") {
 				return err
 			}
 		}
-		if err := api.DropSelectPermissions(views[i], "user"); err != nil {
+		if err := api.DropSelectPermissions(ctx, views[i], "user"); err != nil {
 			log.Warn(err)
 		}
-		if err := api.CreateSelectPermissions(views[i], "user", Permission{
+		if err := api.CreateSelectPermissions(ctx, views[i], "user", Permission{
 			Limit:     hasura.RowsLimit,
 			AllowAggs: hasura.EnableAggregations,
 			Columns:   Columns{"*"},

node/errors.go

@@ -1,8 +1,20 @@
 package node
 
-import "github.com/pkg/errors"
-
-// errors
-var (
-	ErrInvalidResponse = errors.New("invalid response")
+import (
+	"fmt"
 )
+
+// RequestError -
+type RequestError struct {
+	Code int
+	Body string
+	Err  error
+}
+
+// Error -
+func (e RequestError) Error() string {
+	if e.Err != nil {
+		return fmt.Sprintf("%s | %s | status code: %d", e.Err.Error(), e.Body, e.Code)
+	}
+	return fmt.Sprintf("%s | status code: %d", e.Body, e.Code)
+}