Merge remote-tracking branch 'origin/main' into github_auto_repo

.github/workflows/go-lint.yml

@@ -22,7 +22,7 @@ jobs:
         uses: golangci/golangci-lint-action@v3
         with:
           # Optional: version of golangci-lint to use in form of v1.2 or v1.2.3 or `latest` to use the latest version
-          version: v1.56.0
+          version: v1.57.2
 
           # Optional: working directory, useful for monorepos
           # working-directory: somedir

.github/workflows/rill-ui.yml

@@ -43,7 +43,7 @@ jobs:
       - name: Set up NodeJS
         uses: actions/setup-node@v3
         with:
-          node-version: 16
+          node-version: 20
 
       - name: Setup Env variables from Inputs for Prod
         if: ( github.event_name == 'create' && startsWith(github.ref, 'refs/tags/v') ) || ( github.event_name == 'workflow_dispatch' && inputs.env == 'prod' )

.golangci.yml

@@ -95,8 +95,6 @@ linters-settings:
       - filepathJoin
       - hugeParam
       - octalLiteral
-      # Pending https://github.com/go-critic/go-critic/pull/1281
-      - timeCmpSimplify
   gomnd:
     settings:
       mnd:

admin/metrics/client.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/url"
 	"path"
+	"strings"
 	"time"
 )
 
@@ -38,7 +39,18 @@ type AutoscalerSlotsRecommendation struct {
 // AutoscalerSlotsRecommendations invokes the "autoscaler-slots-recommendations" API endpoint to get a list of recommendations for the number of slots to use for projects.
 func (c *Client) AutoscalerSlotsRecommendations(ctx context.Context, limit, offset int) ([]AutoscalerSlotsRecommendation, error) {
 	// Create the URL for the request
-	uri, err := url.Parse(c.RuntimeHost)
+	var runtimeHost string
+
+	// In production, the REST and gRPC endpoints are the same, but in development, they're served on different ports.
+	// TODO: move to http and grpc to the same c.RuntimeHost for local development.
+	// Until we make that change, this is a convenient hack for local development (assumes REST on port 8081).
+	if strings.Contains(c.RuntimeHost, "localhost") {
+		runtimeHost = "http://localhost:8081"
+	} else {
+		runtimeHost = c.RuntimeHost
+	}
+
+	uri, err := url.Parse(runtimeHost)
 	if err != nil {
 		return nil, err
 	}

admin/server/alerts.go

@@ -18,7 +18,9 @@ import (
 	adminv1 "github.com/rilldata/rill/proto/gen/rill/admin/v1"
 	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
 	"github.com/rilldata/rill/runtime"
+	"github.com/rilldata/rill/runtime/drivers/slack"
 	"github.com/rilldata/rill/runtime/pkg/observability"
+	"github.com/rilldata/rill/runtime/pkg/pbutil"
 	"go.opentelemetry.io/otel/attribute"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -274,12 +276,22 @@ func (s *Server) UnsubscribeAlert(ctx context.Context, req *adminv1.UnsubscribeA
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
-	opts := recreateAlertOptionsFromSpec(spec)
+	opts, err := recreateAlertOptionsFromSpec(spec)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to recreate alert options: %s", err.Error())
+	}
 
 	found := false
-	for idx, email := range opts.Recipients {
+	for idx, email := range opts.EmailRecipients {
+		if strings.EqualFold(user.Email, email) {
+			opts.EmailRecipients = slices.Delete(opts.EmailRecipients, idx, idx+1)
+			found = true
+			break
+		}
+	}
+	for idx, email := range opts.SlackUsers {
 		if strings.EqualFold(user.Email, email) {
-			opts.Recipients = slices.Delete(opts.Recipients, idx, idx+1)
+			opts.SlackUsers = slices.Delete(opts.SlackUsers, idx, idx+1)
 			found = true
 			break
 		}
@@ -289,7 +301,7 @@ func (s *Server) UnsubscribeAlert(ctx context.Context, req *adminv1.UnsubscribeA
 		return nil, status.Error(codes.InvalidArgument, "user is not subscribed to alert")
 	}
 
-	if len(opts.Recipients) == 0 {
+	if len(opts.EmailRecipients) == 0 && len(opts.SlackUsers) == 0 && len(opts.SlackChannels) == 0 && len(opts.SlackWebhooks) == 0 {
 		err = s.admin.DB.UpdateVirtualFileDeleted(ctx, proj.ID, proj.ProdBranch, virtualFilePathForManagedAlert(req.Name))
 		if err != nil {
 			return nil, status.Errorf(codes.Internal, "failed to update virtual file: %s", err.Error())
@@ -449,9 +461,13 @@ func (s *Server) yamlForManagedAlert(opts *adminv1.AlertOptions, ownerUserID str
 	res.Query.ArgsJSON = opts.QueryArgsJson
 	// Hard code the user id to run for (to avoid exposing data through alert creation)
 	res.Query.For.UserID = ownerUserID
-	res.Email.Recipients = opts.Recipients
-	res.Email.Renotify = opts.EmailRenotify
-	res.Email.RenotifyAfter = opts.EmailRenotifyAfterSeconds
+	// Notification options
+	res.Renotify = opts.Renotify
+	res.RenotifyAfter = opts.RenotifyAfterSeconds
+	res.Notify.Email.Recipients = opts.EmailRecipients
+	res.Notify.Slack.Channels = opts.SlackChannels
+	res.Notify.Slack.Users = opts.SlackUsers
+	res.Notify.Slack.Webhooks = opts.SlackWebhooks
 	res.Annotations.AdminOwnerUserID = ownerUserID
 	res.Annotations.AdminManaged = true
 	res.Annotations.AdminNonce = time.Now().Format(time.RFC3339Nano)
@@ -477,9 +493,13 @@ func (s *Server) yamlForCommittedAlert(opts *adminv1.AlertOptions) ([]byte, erro
 	res.Intervals.Duration = opts.IntervalDuration
 	res.Query.Name = opts.QueryName
 	res.Query.Args = args
-	res.Email.Recipients = opts.Recipients
-	res.Email.Renotify = opts.EmailRenotify
-	res.Email.RenotifyAfter = opts.EmailRenotifyAfterSeconds
+	// Notification options
+	res.Renotify = opts.Renotify
+	res.RenotifyAfter = opts.RenotifyAfterSeconds
+	res.Notify.Email.Recipients = opts.EmailRecipients
+	res.Notify.Slack.Channels = opts.SlackChannels
+	res.Notify.Slack.Users = opts.SlackUsers
+	res.Notify.Slack.Webhooks = opts.SlackWebhooks
 	return yaml.Marshal(res)
 }
 
@@ -521,16 +541,31 @@ func randomAlertName(title string) string {
 	return name
 }
 
-func recreateAlertOptionsFromSpec(spec *runtimev1.AlertSpec) *adminv1.AlertOptions {
+func recreateAlertOptionsFromSpec(spec *runtimev1.AlertSpec) (*adminv1.AlertOptions, error) {
 	opts := &adminv1.AlertOptions{}
 	opts.Title = spec.Title
 	opts.IntervalDuration = spec.IntervalsIsoDuration
 	opts.QueryName = spec.QueryName
 	opts.QueryArgsJson = spec.QueryArgsJson
-	opts.Recipients = spec.EmailRecipients
-	opts.EmailRenotify = spec.EmailRenotify
-	opts.EmailRenotifyAfterSeconds = spec.EmailRenotifyAfterSeconds
-	return opts
+	opts.Renotify = spec.Renotify
+	opts.RenotifyAfterSeconds = spec.RenotifyAfterSeconds
+	for _, notifier := range spec.Notifiers {
+		switch notifier.Connector {
+		case "email":
+			opts.EmailRecipients = pbutil.ToSliceString(notifier.Properties.AsMap()["recipients"])
+		case "slack":
+			props, err := slack.DecodeProps(notifier.Properties.AsMap())
+			if err != nil {
+				return nil, err
+			}
+			opts.SlackUsers = props.Users
+			opts.SlackChannels = props.Channels
+			opts.SlackWebhooks = props.Webhooks
+		default:
+			return nil, fmt.Errorf("unknown notifier connector: %s", notifier.Connector)
+		}
+	}
+	return opts, nil
 }
 
 // alertYAML is derived from rillv1.AlertYAML, but adapted for generating (as opposed to parsing) the alert YAML.
@@ -550,11 +585,18 @@ type alertYAML struct {
 			UserID string `yaml:"user_id"`
 		} `yaml:"for"`
 	} `yaml:"query"`
-	Email struct {
-		Recipients    []string `yaml:"recipients"`
-		Renotify      bool     `yaml:"renotify"`
-		RenotifyAfter uint32   `yaml:"renotify_after"`
-	} `yaml:"email"`
+	Renotify      bool   `yaml:"renotify"`
+	RenotifyAfter uint32 `yaml:"renotify_after"`
+	Notify        struct {
+		Email struct {
+			Recipients []string `yaml:"emails"`
+		}
+		Slack struct {
+			Users    []string `yaml:"users"`
+			Channels []string `yaml:"channels"`
+			Webhooks []string `yaml:"webhooks"`
+		}
+	}
 	Annotations alertAnnotations `yaml:"annotations,omitempty"`
 }
 

admin/server/reports.go

@@ -17,7 +17,9 @@ import (
 	adminv1 "github.com/rilldata/rill/proto/gen/rill/admin/v1"
 	runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
 	"github.com/rilldata/rill/runtime"
+	"github.com/rilldata/rill/runtime/drivers/slack"
 	"github.com/rilldata/rill/runtime/pkg/observability"
+	"github.com/rilldata/rill/runtime/pkg/pbutil"
 	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/exp/slices"
 	"google.golang.org/grpc/codes"
@@ -246,12 +248,22 @@ func (s *Server) UnsubscribeReport(ctx context.Context, req *adminv1.Unsubscribe
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
-	opts := recreateReportOptionsFromSpec(spec)
+	opts, err := recreateReportOptionsFromSpec(spec)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to recreate report options: %s", err.Error())
+	}
 
 	found := false
-	for idx, email := range opts.Recipients {
+	for idx, email := range opts.EmailRecipients {
+		if strings.EqualFold(user.Email, email) {
+			opts.EmailRecipients = slices.Delete(opts.EmailRecipients, idx, idx+1)
+			found = true
+			break
+		}
+	}
+	for idx, email := range opts.SlackUsers {
 		if strings.EqualFold(user.Email, email) {
-			opts.Recipients = slices.Delete(opts.Recipients, idx, idx+1)
+			opts.SlackUsers = slices.Delete(opts.SlackUsers, idx, idx+1)
 			found = true
 			break
 		}
@@ -261,7 +273,7 @@ func (s *Server) UnsubscribeReport(ctx context.Context, req *adminv1.Unsubscribe
 		return nil, status.Error(codes.InvalidArgument, "user is not subscribed to report")
 	}
 
-	if len(opts.Recipients) == 0 {
+	if len(opts.EmailRecipients) == 0 && len(opts.SlackUsers) == 0 && len(opts.SlackChannels) == 0 && len(opts.SlackWebhooks) == 0 {
 		err = s.admin.DB.UpdateVirtualFileDeleted(ctx, proj.ID, proj.ProdBranch, virtualFilePathForManagedReport(req.Name))
 		if err != nil {
 			return nil, status.Errorf(codes.Internal, "failed to update virtual file: %s", err.Error())
@@ -430,7 +442,10 @@ func (s *Server) yamlForManagedReport(opts *adminv1.ReportOptions, ownerUserID s
 	res.Query.ArgsJSON = opts.QueryArgsJson
 	res.Export.Format = opts.ExportFormat.String()
 	res.Export.Limit = uint(opts.ExportLimit)
-	res.Email.Recipients = opts.Recipients
+	res.Notify.Email.Recipients = opts.EmailRecipients
+	res.Notify.Slack.Channels = opts.SlackChannels
+	res.Notify.Slack.Users = opts.SlackUsers
+	res.Notify.Slack.Webhooks = opts.SlackWebhooks
 	res.Annotations.AdminOwnerUserID = ownerUserID
 	res.Annotations.AdminManaged = true
 	res.Annotations.AdminNonce = time.Now().Format(time.RFC3339Nano)
@@ -470,7 +485,10 @@ func (s *Server) yamlForCommittedReport(opts *adminv1.ReportOptions) ([]byte, er
 	res.Query.Args = args
 	res.Export.Format = exportFormat
 	res.Export.Limit = uint(opts.ExportLimit)
-	res.Email.Recipients = opts.Recipients
+	res.Notify.Email.Recipients = opts.EmailRecipients
+	res.Notify.Slack.Channels = opts.SlackChannels
+	res.Notify.Slack.Users = opts.SlackUsers
+	res.Notify.Slack.Webhooks = opts.SlackWebhooks
 	res.Annotations.WebOpenProjectSubpath = opts.OpenProjectSubpath
 	return yaml.Marshal(res)
 }
@@ -513,7 +531,7 @@ func randomReportName(title string) string {
 	return name
 }
 
-func recreateReportOptionsFromSpec(spec *runtimev1.ReportSpec) *adminv1.ReportOptions {
+func recreateReportOptionsFromSpec(spec *runtimev1.ReportSpec) (*adminv1.ReportOptions, error) {
 	annotations := parseReportAnnotations(spec.Annotations)
 
 	opts := &adminv1.ReportOptions{}
@@ -527,8 +545,23 @@ func recreateReportOptionsFromSpec(spec *runtimev1.ReportSpec) *adminv1.ReportOp
 	opts.ExportLimit = spec.ExportLimit
 	opts.ExportFormat = spec.ExportFormat
 	opts.OpenProjectSubpath = annotations.WebOpenProjectSubpath
-	opts.Recipients = spec.EmailRecipients
-	return opts
+	for _, notifier := range spec.Notifiers {
+		switch notifier.Connector {
+		case "email":
+			opts.EmailRecipients = pbutil.ToSliceString(notifier.Properties.AsMap()["recipients"])
+		case "slack":
+			props, err := slack.DecodeProps(notifier.Properties.AsMap())
+			if err != nil {
+				return nil, err
+			}
+			opts.SlackUsers = props.Users
+			opts.SlackChannels = props.Channels
+			opts.SlackWebhooks = props.Webhooks
+		default:
+			return nil, fmt.Errorf("unknown notifier connector: %s", notifier.Connector)
+		}
+	}
+	return opts, nil
 }
 
 // reportYAML is derived from rillv1.ReportYAML, but adapted for generating (as opposed to parsing) the report YAML.
@@ -548,9 +581,16 @@ type reportYAML struct {
 		Format string `yaml:"format"`
 		Limit  uint   `yaml:"limit"`
 	} `yaml:"export"`
-	Email struct {
-		Recipients []string `yaml:"recipients"`
-	} `yaml:"email"`
+	Notify struct {
+		Email struct {
+			Recipients []string `yaml:"recipients"`
+		} `yaml:"email"`
+		Slack struct {
+			Users    []string `yaml:"users"`
+			Channels []string `yaml:"channels"`
+			Webhooks []string `yaml:"webhooks"`
+		} `yaml:"slack"`
+	} `yaml:"notify"`
 	Annotations reportAnnotations `yaml:"annotations,omitempty"`
 }
 

admin/server/server.go

@@ -169,6 +169,7 @@ func (s *Server) ServeGRPC(ctx context.Context) error {
 
 	adminv1.RegisterAdminServiceServer(server, s)
 	adminv1.RegisterAIServiceServer(server, s)
+	adminv1.RegisterTelemetryServiceServer(server, s)
 	s.logger.Sugar().Infof("serving admin gRPC on port:%v", s.opts.GRPCPort)
 	return graceful.ServeGRPC(ctx, server, s.opts.GRPCPort)
 }
@@ -205,6 +206,10 @@ func (s *Server) HTTPHandler(ctx context.Context) (http.Handler, error) {
 	if err != nil {
 		return nil, err
 	}
+	err = adminv1.RegisterTelemetryServiceHandlerFromEndpoint(ctx, gwMux, grpcAddress, opts)
+	if err != nil {
+		return nil, err
+	}
 
 	// Create regular http mux and mount gwMux on it
 	mux := http.NewServeMux()