Dan/bump pq

.bldr.toml

@@ -1126,11 +1126,7 @@ paths = [
   "lib/io/*",
   "lib/logger/*",
   "lib/pgutils/*",
-  "lib/platform/command/*",
   "lib/platform/config/*",
-  "lib/platform/pg/*",
-  "lib/platform/sys/*",
-  "lib/proc/*",
   "lib/proxy/*",
   "lib/secrets/*",
   "lib/simpledatemath/*",
@@ -1142,7 +1138,6 @@ paths = [
   "vendor/github.com/apache/thrift/*",
   "vendor/github.com/beorn7/perks/*",
   "vendor/github.com/codahale/hdrhistogram/*",
-  "vendor/github.com/davecgh/go-spew/*",
   "vendor/github.com/fsnotify/fsnotify/*",
   "vendor/github.com/go-gorp/gorp/*",
   "vendor/github.com/gogo/protobuf/*",
@@ -1165,7 +1160,6 @@ paths = [
   "vendor/github.com/pelletier/go-toml/*",
   "vendor/github.com/peterbourgon/mergemap/*",
   "vendor/github.com/pkg/errors/*",
-  "vendor/github.com/pmezard/go-difflib/*",
   "vendor/github.com/prometheus/client_golang/*",
   "vendor/github.com/prometheus/client_model/*",
   "vendor/github.com/prometheus/common/*",
@@ -1177,8 +1171,6 @@ paths = [
   "vendor/github.com/spf13/jwalterweatherman/*",
   "vendor/github.com/spf13/pflag/*",
   "vendor/github.com/spf13/viper/*",
-  "vendor/github.com/stretchr/objx/*",
-  "vendor/github.com/stretchr/testify/*",
   "vendor/github.com/teambition/rrule-go/*",
   "vendor/github.com/uber/jaeger-client-go/*",
   "vendor/github.com/uber/jaeger-lib/*",

api/config/pg_gateway/config_request.go

@@ -27,6 +27,7 @@ func DefaultConfigRequest() *ConfigRequest {
 	c.V1.Sys.Service.Port = w.Int32(10145)
 
 	c.V1.Sys.Timeouts.Connect = w.Int32(5)
+	c.V1.Sys.Timeouts.Idle = w.Int32(43200)
 	return c
 }
 

api/config/pg_gateway/config_request.proto

@@ -32,6 +32,7 @@ message ConfigRequest {
 
 			message Timeouts {
 				google.protobuf.Int32Value connect = 1;
+				google.protobuf.Int32Value idle = 2;
 			}
 		}
 	}

components/applications-service/pkg/storage/postgres/services.go

@@ -6,12 +6,11 @@ import (
 	"strings"
 	"time"
 
-	_ "github.com/lib/pq"
+	"github.com/lib/pq"
 	"github.com/pkg/errors"
 
 	"github.com/chef/automate/components/applications-service/pkg/storage"
 	"github.com/chef/automate/lib/pgutils"
-	"github.com/chef/automate/lib/platform/pg"
 )
 
 // composedService is a more user friendly and clear representation of a service.
@@ -303,7 +302,7 @@ func (db *Postgres) GetServicesDistinctValues(fieldName, queryFragment string, f
 		return nil, errors.Errorf("field name %q is not valid for filtering, valid values are %v", fieldName, validFilterFields)
 	}
 
-	columnName := pg.QuoteIdentifier(columnNameForField(fieldName))
+	columnName := pq.QuoteIdentifier(columnNameForField(fieldName))
 	queryFirst := fmt.Sprintf("SELECT DISTINCT %[1]s from service_full AS t WHERE t.%[1]s ILIKE $1 ",
 		columnName,
 	)

components/automate-cli/pkg/diagnostics/context.go

@@ -198,7 +198,7 @@ func (c *testContext) DoLBRequest(path string, opts ...lbrequest.Opts) (*http.Re
 			lastErr = err
 			continue
 		}
-		if resp.StatusCode >= 500 {
+		if resp.StatusCode >= 500 || resp.StatusCode == 403 {
 			lastErr = errors.Errorf("Got 5xx. %d %s: %s", resp.StatusCode, resp.Status, path)
 			resp.Body.Close() // nolint: errcheck
 			continue
@@ -318,20 +318,28 @@ func (c *testContext) PublishViaNATS(messages [][]byte) error {
 }
 
 func (c *testContext) adminToken() (string, error) {
+	if c.Globals.CachedToken != "" {
+		return c.Globals.CachedToken, nil
+	}
+
 	if c.dsClient == nil {
 		return "", ErrDeploymentServiceUnavailable
 	}
 
-	if c.Globals.CachedToken == "" {
-		resp, err := c.dsClient.GenerateAdminToken(context.TODO(), &api.GenerateAdminTokenRequest{
+	var err error
+	var resp *api.GenerateAdminTokenResponse
+
+	for tries := 0; tries < 3; tries++ {
+		time.Sleep(time.Duration(tries) * time.Second)
+		resp, err = c.dsClient.GenerateAdminToken(context.TODO(), &api.GenerateAdminTokenRequest{
 			Description: "This token was generated by the chef-automate diagnostic tool. " +
 				"It has admin level access on the entire Automate API.",
 		})
 
-		if err != nil {
-			return "", err
+		if err == nil {
+			c.Globals.CachedToken = resp.ApiToken
+			break
 		}
-		c.Globals.CachedToken = resp.ApiToken
 	}
 
 	return c.Globals.CachedToken, nil

components/automate-cli/pkg/diagnostics/integration/cfgmgmt_actions.go

@@ -83,15 +83,16 @@ func CreateCfgmgmtActionsDiagnostic() diagnostics.Diagnostic {
 					return err
 				}
 
-				save.CreatedEntities = append(save.CreatedEntities, cfgmgmtActionsEntity{
+				e := cfgmgmtActionsEntity{
 					EntityName:       entityName,
 					RecordedAtMillis: day.UnixNano() / int64(time.Millisecond),
-				})
+				}
+				save.CreatedEntities = append(save.CreatedEntities, e)
 
 				tstCtx.SetValue("cfgmgmt-actions", save)
 
 				resp, err := tstCtx.DoLBRequest(
-					"/api/v0/events/data-collector",
+					"/api/v0/events/data-collector?z=cfgmgmt-actions",
 					lbrequest.WithMethod("POST"),
 					lbrequest.WithJSONBody(buf.String()),
 				)
@@ -103,7 +104,7 @@ func CreateCfgmgmtActionsDiagnostic() diagnostics.Diagnostic {
 					_ = resp.Body.Close()
 				}()
 				if resp.StatusCode != 200 {
-					return errors.Wrapf(err, "Failed to POST /api/v0/events/data-collector: Body:\n%s", buf.String())
+					return errors.Errorf("Failed to POST /api/v0/events/data-collector: Body:\n%s", buf.String())
 				}
 			}
 
@@ -125,6 +126,15 @@ func CreateCfgmgmtActionsDiagnostic() diagnostics.Diagnostic {
 			for _, entity := range loaded.CreatedEntities {
 				reqPath := fmt.Sprintf("/api/v0/eventfeed?collapse=true&page_size=100&start=%d&end=%d", entity.RecordedAtMillis-1, entity.RecordedAtMillis+1)
 				found := false
+
+				type eventsFeedResp struct {
+					Events []struct {
+						EntityName string `json:"entity_name"`
+					} `json:"events"`
+				}
+
+				respUnmarshalled := eventsFeedResp{}
+
 			RETRY_LOOP:
 				for {
 					resp, err := tstCtx.DoLBRequest(reqPath)
@@ -137,13 +147,7 @@ func CreateCfgmgmtActionsDiagnostic() diagnostics.Diagnostic {
 					// We wont retry a flakey backend. It should always return 200
 					require.Equal(tstCtx, 200, resp.StatusCode, "Failed to GET %s", reqPath)
 
-					type eventsFeedResp struct {
-						Events []struct {
-							EntityName string `json:"entity_name"`
-						} `json:"events"`
-					}
-
-					respUnmarshalled := eventsFeedResp{}
+					respUnmarshalled = eventsFeedResp{}
 					err = json.NewDecoder(resp.Body).Decode(&respUnmarshalled)
 					// We should always get valid json
 					require.NoError(tstCtx, err, "Failed to decode body of GET %s", reqPath)
@@ -158,9 +162,9 @@ func CreateCfgmgmtActionsDiagnostic() diagnostics.Diagnostic {
 					if tries >= maxTries {
 						break RETRY_LOOP
 					}
-					time.Sleep(2 * time.Duration(tries) * time.Second)
+					time.Sleep(5 * time.Duration(tries) * time.Second)
 				}
-				assert.True(tstCtx, found, "Could not find entity %s in GET %s", entity.EntityName, reqPath)
+				assert.True(tstCtx, found, "[%s] Could not find entity %s in GET %s; did get %+v", time.Now().UTC().String(), entity.EntityName, reqPath, respUnmarshalled)
 			}
 
 		},

components/automate-pg-gateway/habitat/config/haproxy.conf

@@ -11,7 +11,7 @@ defaults
     # how long a long-lived connection is allowed to stayl idle.  We
     # likely want to keep this high to avoid unnecessarily killing
     # long-lived SQL connections used by low-request-rate services.
-    timeout tunnel 300s
+    timeout tunnel {{cfg.timeouts.idle}}s
     # Client inactivity connections for half-closed connections (i.e
     # the server has closed the connection and but the client has
     # not). The HAProxy documentation recommends setting this whenever

components/automate-pg-gateway/habitat/default.toml

@@ -5,3 +5,5 @@ status_port = 10146
 
 [timeouts]
 connect = 5
+# 12 hours
+idle = 43200

components/ingest-service/pipeline/message/chef_action.go

@@ -42,12 +42,15 @@ func (chefAction ChefAction) FinishProcessing(err error) {
 	if err == nil {
 		// Adding new metric; Time for a message to go through the pipeline
 		log.WithFields(log.Fields{
-			"message_id": chefAction.ID,
-			"message":    "ChefAction",
-			"metric":     "pipeline",
-			"type":       "ingest_time",
-			"ms":         chefAction.ClockProcessingTime(),
-		}).Info("Message ingested successfully")
+			"message_id":  chefAction.ID,
+			"message":     "ChefAction",
+			"metric":      "pipeline",
+			"type":        "ingest_time",
+			"ms":          chefAction.ClockProcessingTime(),
+			"entity_name": chefAction.InternalChefAction.EntityName,
+			"entity_type": chefAction.InternalChefAction.EntityType,
+			"entity_task": chefAction.InternalChefAction.Task,
+		}).Info("Chef Action message ingested successfully")
 	}
 }
 

components/ingest-service/pipeline/publisher/chef_action.go

@@ -34,6 +34,14 @@ func ChefAction(in <-chan message.ChefAction, client backend.Client, out chan<-
 
 			err := client.InsertAction(msg.Ctx, msg.InternalChefAction)
 			if err != nil {
+				log.WithError(err).WithFields(log.Fields{
+					"publisher_id": number,
+					"message_id":   msg.ID,
+					"buffer_size":  len(out),
+					"entity_name":  msg.InternalChefAction.EntityName,
+					"entity_type":  msg.InternalChefAction.EntityType,
+					"entity_task":  msg.InternalChefAction.Task,
+				}).Error("Failed to insert Chef Action")
 				msg.FinishProcessing(status.Errorf(codes.Internal, err.Error()))
 			} else {
 				out <- msg

components/ingest-service/server/chef.go

@@ -65,7 +65,7 @@ func (s *ChefIngestServer) ProcessChefRun(ctx context.Context, run *chef.Run) (*
 		err = <-errc
 
 		if err != nil {
-			log.WithError(err).Error("Message failure")
+			log.WithError(err).Error("Chef run ingestion failure")
 		}
 	} else if run.GetMessageType() == "run_start" {
 		log.WithFields(log.Fields{
@@ -95,7 +95,7 @@ func (s *ChefIngestServer) ProcessChefAction(ctx context.Context, action *chef.A
 		err := <-errc
 
 		if err != nil {
-			log.WithError(err).Error("Message failure")
+			log.WithError(err).Error("Chef Action ingestion failure")
 		}
 		return &response.ProcessChefActionResponse{}, err
 	}

components/pg-sidecar-service/integration/pg_sidecar_test.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/gofrs/uuid"
+	"github.com/lib/pq"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/status"
@@ -578,7 +579,7 @@ SELECT EXISTS
   )
 ;`
 
-	return c.BoolQuery(fmt.Sprintf(tableOwnedByQuery, pg.QuoteLiteral(role), pg.QuoteLiteral(table)))
+	return c.BoolQuery(fmt.Sprintf(tableOwnedByQuery, pq.QuoteLiteral(role), pq.QuoteLiteral(table)))
 }
 
 func (h *pgHelper) hasSuperuser(role string) (bool, error) {

components/pg-sidecar-service/pkg/pgw/alter_role_query.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/chef/automate/lib/platform/pg"
+	"github.com/lib/pq"
 )
 
 // AlterRoleQuery represents the available options for an ALTER ROLE query
@@ -94,12 +94,12 @@ func (o *AlterRoleQuery) String() string {
 		if o.Password.Unencrypted {
 			options = append(options, "UNENCRYPTED")
 		}
-		options = append(options, fmt.Sprintf("PASSWORD %s", pg.QuoteLiteral(o.Password.Value)))
+		options = append(options, fmt.Sprintf("PASSWORD %s", pq.QuoteLiteral(o.Password.Value)))
 	}
 
 	if len(options) == 0 {
 		return ""
 	}
 
-	return fmt.Sprintf("ALTER ROLE %s WITH %s", pg.QuoteIdentifier(o.Role), strings.Join(options, " "))
+	return fmt.Sprintf("ALTER ROLE %s WITH %s", pq.QuoteIdentifier(o.Role), strings.Join(options, " "))
 }