obsservice: connect to the sink cluster and run migrations

go.mod

@@ -93,7 +93,7 @@ require (
 	github.com/jackc/pgconn v1.12.1
 	github.com/jackc/pgproto3/v2 v2.3.0
 	github.com/jackc/pgtype v1.11.0
-	github.com/jackc/pgx/v4 v4.16.0
+	github.com/jackc/pgx/v4 v4.16.1
 	github.com/jaegertracing/jaeger v1.18.1
 	github.com/jordan-wright/email v4.0.1-0.20210109023952-943e75fe5223+incompatible
 	github.com/jordanlewis/gcassert v0.0.0-20210709222130-81f5df3faab8
@@ -119,12 +119,13 @@ require (
 	github.com/mmatczuk/go_generics v0.0.0-20181212143635-0aaa050f9bab
 	github.com/montanaflynn/stats v0.6.3
 	github.com/olekukonko/tablewriter v0.0.5-0.20200416053754-163badb3bac6
-	github.com/opencontainers/image-spec v1.0.1
+	github.com/opencontainers/image-spec v1.0.2
 	github.com/otan/gopgkrb5 v1.0.3
 	github.com/petermattis/goid v0.0.0-20211229010228-4d14c490ee36
 	github.com/pierrre/geohash v1.0.0
 	github.com/pkg/browser v0.0.0-20180916011732-0a3d74bf9ce4
 	github.com/pmezard/go-difflib v1.0.0
+	github.com/pressly/goose/v3 v3.5.3
 	github.com/prometheus/client_golang v1.12.0
 	github.com/prometheus/client_model v0.2.1-0.20210607210712-147c58e9608a
 	github.com/prometheus/common v0.32.1
@@ -163,7 +164,7 @@ require (
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211
 	golang.org/x/text v0.3.7
 	golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac
-	golang.org/x/tools v0.1.8-0.20211029000441-d6a9af8af023
+	golang.org/x/tools v0.1.9
 	google.golang.org/api v0.80.0
 	google.golang.org/genproto v0.0.0-20220505152158-f39f71e6c8f3
 	google.golang.org/grpc v1.46.0
@@ -179,7 +180,7 @@ require (
 	cloud.google.com/go/compute v1.6.1 // indirect
 	cloud.google.com/go/iam v0.3.0 // indirect
 	github.com/Azure/azure-pipeline-go v0.2.3 // indirect
-	github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/Azure/go-autorest v14.2.0+incompatible // indirect
 	github.com/Azure/go-autorest/autorest/adal v0.9.15 // indirect
 	github.com/Azure/go-autorest/autorest/date v0.3.0 // indirect
@@ -190,7 +191,7 @@ require (
 	github.com/Masterminds/goutils v1.1.0 // indirect
 	github.com/Masterminds/semver v1.5.0 // indirect
 	github.com/Masterminds/sprig v2.22.0+incompatible // indirect
-	github.com/Microsoft/go-winio v0.4.17 // indirect
+	github.com/Microsoft/go-winio v0.5.1 // indirect
 	github.com/PuerkitoBio/purell v1.1.1 // indirect
 	github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578 // indirect
 	github.com/abbot/go-http-auth v0.4.1-0.20181019201920-860ed7f246ff // indirect
@@ -208,7 +209,7 @@ require (
 	github.com/aws/smithy-go v1.11.2 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bgentry/go-netrc v0.0.0-20140422174119-9fd32a8b3d3d // indirect
-	github.com/cenkalti/backoff/v4 v4.1.1 // indirect
+	github.com/cenkalti/backoff/v4 v4.1.2 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dimchansky/utfbom v1.1.1 // indirect
@@ -251,7 +252,7 @@ require (
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/huandu/xstrings v1.3.0 // indirect
 	github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639 // indirect
-	github.com/imdario/mergo v0.3.11 // indirect
+	github.com/imdario/mergo v0.3.12 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
 	github.com/jackc/pgio v1.0.0 // indirect

pkg/obsservice/README.md

@@ -52,6 +52,10 @@ obsservice --http-addr=localhost:8081 --crdb-http-url=http://localhost:8080 --ui
   one created with `cockroach cert create-ca`). If specified, HTTP requests are
   only proxied to CRDB nodes that present certificates signed by this CA. If not
   specified, the system's CA list is used.
+- `--sink-pgurl` is the connection string for the sink cluster. If the pgurl
+  contains a database name, that database will be used; otherwise `obsservice`
+  will be used. If not specified, a connection to a local cluster will be
+  attempted.
 
 ## Functionality
 
@@ -63,6 +67,11 @@ CRDB. Currently the Obs Service doesn't handle any routes (other than
 `/debug/pprof/*`, which exposes the Obs Service's own pprof endpoints), so all
 requests are forwarded.
 
+The Obs Service connects to a sink cluster identified by `--sink-pgurl`. The
+required schema is automatically created using SQL migrations run with
+[goose](https://github.com/pressly/goose). The state of migrations in a sink
+cluster can be inspected through the `observice.obs_admin.migrations` table.
+
 ## Licensing
 
 The Observability Service is licensed as Apache 2.0.

pkg/obsservice/cmd/obsservice/BUILD.bazel

@@ -7,7 +7,8 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//pkg/cli/exit",
-        "//pkg/obsservice/obslib",
+        "//pkg/obsservice/obslib/httpproxy",
+        "//pkg/obsservice/obslib/migrations",
         "@com_github_spf13_cobra//:cobra",
     ],
 )

pkg/obsservice/cmd/obsservice/main.go

@@ -10,10 +10,12 @@ package main
 
 import (
 	"context"
+	"flag"
 	"fmt"
 
 	"github.com/cockroachdb/cockroach/pkg/cli/exit"
-	"github.com/cockroachdb/cockroach/pkg/obsservice/obslib"
+	"github.com/cockroachdb/cockroach/pkg/obsservice/obslib/httpproxy"
+	"github.com/cockroachdb/cockroach/pkg/obsservice/obslib/migrations"
 	"github.com/spf13/cobra"
 )
 
@@ -25,16 +27,20 @@ var RootCmd = &cobra.Command{
 from one or more CockroachDB clusters.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		ctx := context.Background()
-		cfg := obslib.ReverseHTTPProxyConfig{
+		cfg := httpproxy.ReverseHTTPProxyConfig{
 			HTTPAddr:      httpAddr,
 			TargetURL:     targetURL,
 			CACertPath:    caCertPath,
 			UICertPath:    uiCertPath,
 			UICertKeyPath: uiCertKeyPath,
 		}
 
+		if err := migrations.RunDBMigrations(ctx, sinkPGURL); err != nil {
+			panic(err)
+		}
+
 		// Block forever running the proxy.
-		<-obslib.NewReverseHTTPProxy(ctx, cfg).RunAsync(ctx)
+		<-httpproxy.NewReverseHTTPProxy(ctx, cfg).RunAsync(ctx)
 	},
 }
 
@@ -44,9 +50,15 @@ var (
 	targetURL                 string
 	caCertPath                string
 	uiCertPath, uiCertKeyPath string
+	sinkPGURL                 string
 )
 
 func main() {
+
+	// Add all the flags registered with the standard "flag" package. Useful for
+	// --vmodule, for example.
+	RootCmd.PersistentFlags().AddGoFlagSet(flag.CommandLine)
+
 	RootCmd.PersistentFlags().StringVar(
 		&httpAddr,
 		"http-addr",
@@ -76,6 +88,13 @@ func main() {
 		"Path to the private key used by the Observability Service. "+
 			"This is the key corresponding to the --ui-cert certificate.")
 
+	// Flags about connecting to the sink cluster.
+	RootCmd.PersistentFlags().StringVar(
+		&sinkPGURL,
+		"sink-pgurl",
+		"postgresql://root@andrei-desktop:26257/defaultdb?sslmode=disable",
+		"PGURL for the sink cluster.")
+
 	if err := RootCmd.Execute(); err != nil {
 		fmt.Println(err)
 		exit.WithCode(exit.UnspecifiedError())

pkg/obsservice/obslib/BUILD.bazel

@@ -1,14 +0,0 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
-
-go_library(
-    name = "obslib",
-    srcs = ["lib.go"],
-    importpath = "github.com/cockroachdb/cockroach/pkg/obsservice/obslib",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/util/log",
-        "//pkg/util/syncutil",
-        "@com_github_cockroachdb_cmux//:cmux",
-        "@com_github_cockroachdb_errors//:errors",
-    ],
-)

pkg/obsservice/obslib/httpproxy/BUILD.bazel

@@ -0,0 +1,15 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "httpproxy",
+    srcs = ["reverseproxy.go"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/obsservice/obslib/httpproxy",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/cli/exit",
+        "//pkg/util/log",
+        "//pkg/util/syncutil",
+        "@com_github_cockroachdb_cmux//:cmux",
+        "@com_github_cockroachdb_errors//:errors",
+    ],
+)

pkg/obsservice/obslib/lib.go → pkg/obsservice/obslib/httpproxy/reverseproxy.go

@@ -6,7 +6,7 @@
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 
-package obslib
+package httpproxy
 
 import (
 	"context"
@@ -20,9 +20,11 @@ import (
 	"net/http/httputil"
 	"net/http/pprof"
 	"net/url"
+	"os"
 	"time"
 
 	"github.com/cockroachdb/cmux"
+	"github.com/cockroachdb/cockroach/pkg/cli/exit"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/errors"
@@ -101,15 +103,18 @@ to trust the certificate presented by CockroachDB.`)
 func (p *ReverseHTTPProxy) RunAsync(ctx context.Context) <-chan struct{} {
 	ch := make(chan struct{})
 
+	listener, err := net.Listen("tcp", p.listenAddr)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "failed to listen for incoming HTTP connections on address %s: %s",
+			p.listenAddr, err)
+		exit.WithCode(exit.UnspecifiedError())
+	}
+
 	https := p.certs.UICert != nil
 	go func() {
 		defer close(ch)
 		var err error
 
-		listener, err := net.Listen("tcp", p.listenAddr)
-		if err != nil {
-			log.Fatalf(ctx, "%s", err)
-		}
 		defer func() {
 			_ = listener.Close()
 		}()

pkg/obsservice/obslib/migrations/BUILD.bazel

@@ -0,0 +1,15 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "migrations",
+    srcs = ["migrations.go"],
+    embedsrcs = ["sqlmigrations/0001_init.sql"],
+    importpath = "github.com/cockroachdb/cockroach/pkg/obsservice/obslib/migrations",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//pkg/util/log",
+        "@com_github_jackc_pgx_v4//:pgx",
+        "@com_github_jackc_pgx_v4//stdlib",
+        "@com_github_pressly_goose_v3//:goose",
+    ],
+)

pkg/obsservice/obslib/migrations/migrations.go

@@ -0,0 +1,69 @@
+// Copyright 2022 The Cockroach Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+
+package migrations
+
+import (
+	"context"
+	"embed"
+	"fmt"
+
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/stdlib"
+	"github.com/pressly/goose/v3"
+)
+
+// sqlMigrations embeds all the .sql file containing migrations to be run by
+// Goose.
+//go:embed sqlmigrations/*.sql
+var sqlMigrations embed.FS
+
+// defaultSinkDBName is the name of the database to be used by default
+const defaultSinkDBName = "obsservice"
+
+// RunDBMigrations brings the SQL schema in the sink cluster up to date.
+//
+// sinkPGURL is the connection string for the sink cluster. If it includes a
+// database, that database will be used. If it doesn't, a default one will be
+// used.
+func RunDBMigrations(ctx context.Context, sinkPGURL string) error {
+	connCfg, err := pgx.ParseConfig(sinkPGURL)
+	if err != nil {
+		return err
+	}
+	if connCfg.Database == "" {
+		fmt.Printf("No database explicitly provided in --sink-pgurl. Using %q.\n", defaultSinkDBName)
+		connCfg.Database = defaultSinkDBName
+	}
+
+	if log.V(2) {
+		goose.SetVerbose(true)
+	}
+	goose.SetBaseFS(sqlMigrations)
+
+	db := stdlib.OpenDB(*connCfg)
+	// We need to create the database by hand; Goose expects the database to exist.
+	if _, err := db.ExecContext(ctx, "CREATE DATABASE IF NOT EXISTS "+connCfg.Database); err != nil {
+		return err
+	}
+	// goose will <db>.obs_admin.migrations to store the migration bookkeeping.
+	if _, err := db.ExecContext(ctx, "CREATE schema IF NOT EXISTS obs_admin"); err != nil {
+		return err
+	}
+	goose.SetTableName("obs_admin.migrations")
+	if err := goose.SetDialect("postgres"); err != nil {
+		return err
+	}
+
+	// Run the missing migrations, if any.
+	if err := goose.Up(db, "sqlmigrations"); err != nil {
+		return err
+	}
+	return nil
+}

pkg/obsservice/obslib/migrations/sqlmigrations/0001_init.sql

@@ -0,0 +1,5 @@
+-- +goose Up
+CREATE TABLE events();
+
+-- +goose Down
+DROP TABLE events;