Add support for running dashboards (#218)

Author: bcspragu

WORKSPACE

@@ -117,3 +117,11 @@ oci_pull(
     image = "ghcr.io/rmi-pacta/workflow.portfolio.parsing",
     platforms = ["linux/amd64"],
 )
+
+oci_pull(
+    name = "dashboard_base",
+    # This digest is of the 'main' tag as of 2024-12-19
+    digest = "sha256:dea705bac105a0847a6070f1914298fb7145273bc4a56b76ba8ec80ce427e269",
+    image = "ghcr.io/rmi-pacta/workflow.pacta.dashboard",
+    platforms = ["linux/amd64"],
+)

async/async.go

@@ -246,27 +246,38 @@ type ReportInputPortfolio struct {
 	Name         string   `json:"name"`
 }
 
-type ReportEnv struct {
+type DashboardInput struct {
+	Portfolio DashboardInputPortfolio `json:"portfolio"`
+	Inherit   string                  `json:"inherit"`
+}
+
+type DashboardInputPortfolio struct {
+	Files        []string `json:"files"`
+	HoldingsDate string   `json:"holdingsDate"`
+	Name         string   `json:"name"`
+}
+
+type TaskEnv struct {
 	rootDir string
 
 	// These are mounted in from externally.
 	benchmarksDir string
 	pactaDataDir  string
 }
 
-func initReportEnv(benchmarkDir, pactaDataDir, baseDir string) (*ReportEnv, error) {
+func initEnv(benchmarkDir, pactaDataDir, baseDir, taskName string) (*TaskEnv, error) {
 	// Make sure the base directory exists first.
 	if err := os.MkdirAll(baseDir, 0700); err != nil {
 		return nil, fmt.Errorf("failed to create base input dir: %w", err)
 	}
 	// We create temp subdirectories, because while this code currently executes in
 	// a new container for each invocation, that might not always be the case.
-	rootDir, err := os.MkdirTemp(baseDir, "create-report")
+	rootDir, err := os.MkdirTemp(baseDir, taskName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create temp dir for input CSVs: %w", err)
 	}
 
-	re := &ReportEnv{
+	re := &TaskEnv{
 		rootDir:       rootDir,
 		benchmarksDir: benchmarkDir,
 		pactaDataDir:  pactaDataDir,
@@ -293,15 +304,15 @@ const (
 	SummaryOutputDir  = ReportDir("summary-output")
 )
 
-func (r *ReportEnv) outputDirs() []string {
+func (r *TaskEnv) outputDirs() []string {
 	return []string{
 		r.pathForDir(AnalysisOutputDir),
 		r.pathForDir(ReportOutputDir),
 		r.pathForDir(SummaryOutputDir),
 	}
 }
 
-func (r *ReportEnv) asEnvVars() []string {
+func (r *TaskEnv) asEnvVars() []string {
 	return []string{
 		"BENCHMARKS_DIR=" + r.benchmarksDir,
 		"PACTA_DATA_DIR=" + r.pactaDataDir,
@@ -315,11 +326,11 @@ func (r *ReportEnv) asEnvVars() []string {
 	}
 }
 
-func (r *ReportEnv) pathForDir(d ReportDir) string {
+func (r *TaskEnv) pathForDir(d ReportDir) string {
 	return filepath.Join(r.rootDir, string(d))
 }
 
-func (r *ReportEnv) makeDirectories() error {
+func (r *TaskEnv) makeDirectories() error {
 	var rErr error
 	makeDir := func(reportDir ReportDir) {
 		if rErr != nil {
@@ -349,6 +360,104 @@ func (r *ReportEnv) makeDirectories() error {
 	return nil
 }
 
+func (h *Handler) CreateDashboard(ctx context.Context, taskID task.ID, req *task.CreateDashboardRequest, dashboardContainer string) error {
+	if n := len(req.BlobURIs); n != 1 {
+		return fmt.Errorf("expected exactly one blob URI as input, got %d", n)
+	}
+	blobURI := req.BlobURIs[0]
+
+	//  We use this instead of /mnt/... because the base image (quite
+	// reasonably) uses a non-root user, so we can't be creating directories in the
+	// root filesystem all willy nilly.
+	baseDir := filepath.Join("/", "home", "workflow-pacta-webapp")
+
+	dashEnv, err := initEnv(h.benchmarkDir, h.pactaDataDir, baseDir, "create-dashboard")
+	if err != nil {
+		return fmt.Errorf("failed to init report env: %w", err)
+	}
+
+	// Load the parsed portfolio from blob storage, place it in our PORFOLIO_DIR,
+	// where the `prepare_dashboard_data.R` script expects it to be.
+	fileNameWithExt := filepath.Base(string(blobURI))
+	if !strings.HasSuffix(fileNameWithExt, ".csv") {
+		return fmt.Errorf("given blob wasn't a CSV-formatted portfolio, %q", fileNameWithExt)
+	}
+	destPath := filepath.Join(dashEnv.pathForDir(PortfoliosDir), fileNameWithExt)
+	if err := h.downloadBlob(ctx, string(blobURI), destPath); err != nil {
+		return fmt.Errorf("failed to download processed portfolio blob: %w", err)
+	}
+
+	inp := DashboardInput{
+		Portfolio: DashboardInputPortfolio{
+			Files:        []string{fileNameWithExt},
+			HoldingsDate: "2023-12-31",   // TODO(#206)
+			Name:         "FooPortfolio", // TODO(#206)
+		},
+		Inherit: "GENERAL_2023Q4", // TODO(#206): Should this be configurable
+	}
+
+	var inpJSON bytes.Buffer
+	if err := json.NewEncoder(&inpJSON).Encode(inp); err != nil {
+		return fmt.Errorf("failed to encode report input as JSON: %w", err)
+	}
+
+	cmd := exec.CommandContext(ctx,
+		"/usr/local/bin/Rscript",
+		"--vanilla", "/workflow.pacta.dashboard/inst/extdata/scripts/prepare_dashboard_data.R",
+		inpJSON.String())
+
+	cmd.Env = append(cmd.Env, dashEnv.asEnvVars()...)
+	cmd.Env = append(cmd.Env,
+		"LOG_LEVEL=DEBUG",
+		"HOME=/root", /* Required by pandoc */
+	)
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+
+	if err := cmd.Run(); err != nil {
+		return fmt.Errorf("failed to run pacta dashboard script: %w", err)
+	}
+
+	var artifacts []*task.AnalysisArtifact
+	uploadDir := func(dir string) error {
+		aas, err := h.uploadDirectory(ctx, dir, dashboardContainer, req.AnalysisID)
+		if err != nil {
+			return fmt.Errorf("failed to upload report directory: %w", err)
+		}
+		artifacts = append(artifacts, aas...)
+		return nil
+	}
+
+	for _, outDir := range dashEnv.outputDirs() {
+		if err := uploadDir(outDir); err != nil {
+			return fmt.Errorf("failed to upload artifacts %q: %w", outDir, err)
+		}
+	}
+
+	events := []publisher.Event{
+		{
+			Data: task.CreateDashboardResponse{
+				TaskID:    taskID,
+				Request:   req,
+				Artifacts: artifacts,
+			},
+			DataVersion: to.Ptr("1.0"),
+			EventType:   to.Ptr("created-dashboard"),
+			EventTime:   to.Ptr(time.Now()),
+			ID:          to.Ptr(string(taskID)),
+			Subject:     to.Ptr(string(taskID)),
+		},
+	}
+
+	if _, err := h.pubsub.PublishEvents(ctx, events, nil); err != nil {
+		return fmt.Errorf("failed to publish event: %w", err)
+	}
+
+	h.logger.Info("created report", zap.String("task_id", string(taskID)))
+
+	return nil
+}
+
 func (h *Handler) CreateReport(ctx context.Context, taskID task.ID, req *task.CreateReportRequest, reportContainer string) error {
 	if n := len(req.BlobURIs); n != 1 {
 		return fmt.Errorf("expected exactly one blob URI as input, got %d", n)
@@ -360,7 +469,7 @@ func (h *Handler) CreateReport(ctx context.Context, taskID task.ID, req *task.Cr
 	// root filesystem all willy nilly.
 	baseDir := filepath.Join("/", "home", "workflow-pacta-webapp")
 
-	reportEnv, err := initReportEnv(h.benchmarkDir, h.pactaDataDir, baseDir)
+	reportEnv, err := initEnv(h.benchmarkDir, h.pactaDataDir, baseDir, "create-report")
 	if err != nil {
 		return fmt.Errorf("failed to init report env: %w", err)
 	}

async/req.go

@@ -14,6 +14,10 @@ func LoadParsePortfolioRequestFromEnv() (*task.ParsePortfolioRequest, error) {
 	return loadFromEnv[task.ParsePortfolioRequest]("PARSE_PORTFOLIO_REQUEST", "ParsePortfolioRequest")
 }
 
+func LoadCreateDashboardRequestFromEnv() (*task.CreateDashboardRequest, error) {
+	return loadFromEnv[task.CreateDashboardRequest]("CREATE_DASHBOARD_REQUEST", "CreateDashboardRequest")
+}
+
 func LoadCreateAuditRequestFromEnv() (*task.CreateAuditRequest, error) {
 	return loadFromEnv[task.CreateAuditRequest]("CREATE_AUDIT_REQUEST", "CreateAuditRequest")
 }

azure/azevents/azevents.go

@@ -276,6 +276,14 @@ func (s *Server) handleEventGrid(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		s.handleCreatedReport(req.ID, &resp, w)
+	case "created-dashboard":
+		var resp task.CreateDashboardResponse
+		if err := json.Unmarshal(req.Data, &resp); err != nil {
+			s.logger.Error("failed to parse event data as CreateDashboardResponse", zap.String("event_grid_id", req.ID), zap.Error(err))
+			http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
+			return
+		}
+		s.handleCreatedDashboard(req.ID, &resp, w)
 	default:
 		s.logger.Error("unexpected event type", zap.String("event_grid_id", req.ID), zap.String("event_type", req.EventType))
 		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
@@ -400,6 +408,15 @@ func (s *Server) handleCreatedReport(id string, resp *task.CreateReportResponse,
 		w)
 }
 
+func (s *Server) handleCreatedDashboard(id string, resp *task.CreateDashboardResponse, w http.ResponseWriter) {
+	s.handleCompletedAnalysis(
+		pacta.AnalysisType_Dashboard,
+		resp.Request.AnalysisID,
+		id,
+		resp.Artifacts,
+		w)
+}
+
 func (s *Server) handleCompletedAnalysis(
 	analysisType pacta.AnalysisType,
 	analysisID pacta.AnalysisID,

cmd/dashboard/BUILD.bazel

@@ -0,0 +1,71 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_binary", "go_library")
+load("@rules_pkg//:pkg.bzl", "pkg_tar")
+load("@rules_oci//oci:defs.bzl", "oci_image", "oci_push", "oci_tarball")
+
+go_library(
+    name = "dashboard_lib",
+    srcs = ["main.go"],
+    importpath = "github.com/RMI/pacta/cmd/dashboard",
+    visibility = ["//visibility:private"],
+    deps = [
+        "//async",
+        "//azure/azblob",
+        "//azure/azcreds",
+        "//azure/azlog",
+        "//task",
+        "@com_github_azure_azure_sdk_for_go_sdk_azidentity//:azidentity",
+        "@com_github_azure_azure_sdk_for_go_sdk_messaging_azeventgrid//publisher",
+        "@com_github_namsral_flag//:flag",
+        "@org_uber_go_zap//:zap",
+        "@org_uber_go_zap//zapcore",
+        "@org_uber_go_zap_exp//zapfield",
+    ],
+)
+
+go_binary(
+    name = "dashboard",
+    embed = [":dashboard_lib"],
+    visibility = ["//visibility:public"],
+)
+
+pkg_tar(
+    name = "dashboard_tar",
+    srcs = [":dashboard"],
+)
+
+filegroup(
+    name = "configs",
+    srcs = glob(["configs/**"]),
+    visibility = ["//visibility:public"],
+)
+
+pkg_tar(
+    name = "configs_tar",
+    srcs = [":configs"],
+    package_dir = "/configs",
+    strip_prefix = "/cmd/dashboard/configs",
+)
+
+oci_image(
+    name = "image",
+    base = "@dashboard_base",
+    entrypoint = ["/dashboard"],
+    tars = [
+        ":dashboard_tar",
+        ":configs_tar",
+    ],
+)
+
+oci_push(
+    name = "push_image",
+    image = ":image",
+    remote_tags = ["latest"],
+    repository = "rmisppactaweupatdev.azurecr.io/pactadashboard",
+)
+
+# Note: This tarball is provided for local testing of the Docker image, see the README.md for details on usage.
+oci_tarball(
+    name = "image_tarball",
+    image = ":image",
+    repo_tags = [],
+)

cmd/dashboard/README.md

@@ -0,0 +1,32 @@
+# Dashboard
+
+This directory contains the `dashboard` binary, which acts as a thin shim around the PACTA [`workflow.pacta.dashboard` tooling](https://github.com/RMI-PACTA/workflow.pacta.dashboard), running tasks created via either Azure Container App Jobs (via the `aztask` package) or local Docker (`dockertask`), loading relevant blobs, and writing relevant outputs.
+
+## Running locally
+
+The `dashboard` binary doesn't need to be run locally in order to test PACTA processing. By default, the backend API server will trigger PACTA dashboard creation runs against a local Docker daemon, testing most of the run-handling code in the process (e.g. file handling, task execution, etc).
+
+If you do want to actually run the full `dashboard` image on Azure, you can use:
+
+```bash
+# Run the backend, tell it to create tasks as real Azure Container Apps Jobs.
+bazel run //scripts:run_server -- --use_azure_runner
+```
+
+### Creating a new docker image to run locally
+
+When developing the runner, you have two options:
+
+* **Test against local Docker** - Run the server **without** the  `--use_azure_runner`, which means async tasks will run locally, using `docker run ...`. To test local runner changes, you can build and tag a runner image locally with `bazel run //scripts:build_and_load_dashboard`.
+  * After running the script, the updated runner will immediately be available, no need to restart the server.
+  * This is the option you'll want to use most of the time.
+* **Test against Azure Container Apps Jobs** - Run the server **with** the  `--use_azure_runner`, which means async tasks will be run on Azure, created via the Azure API. To test changes here, you can build and tag a runner image locally with `bazel run //scripts:build_and_load_dashboard`, and then push it to Azure with `docker push rmisa.azurecr.io/pactadashboard:latest`
+  * You generally won't need to use this option unless you're testing something very specific about the runner's integration with Azure, as the runner code is identical whether run locally or on Azure.
+
+### Cleaning up old dashboard containers
+
+By default, we don't auto-remove stopped containers (i.e. finished dashboard tasks), to give developers a chance to review the logs (e.g. with `docker logs <sha>`). To clean up all completed runs at once, run:
+
+```bash
+docker rm $(docker ps -a -q -f "status=exited" -f "ancestor=rmisa.azurecr.io/pactadashboard:latest")
+```

cmd/dashboard/configs/dev.conf

@@ -0,0 +1,8 @@
+env dev
+min_log_level warn
+
+azure_event_topic pacta-events-dev
+azure_topic_location centralus-1
+
+azure_storage_account rmipactadev
+azure_dest_portfolio_container parsedportfolios

cmd/dashboard/configs/local.conf

@@ -0,0 +1,8 @@
+env local
+min_log_level debug
+
+azure_event_topic pacta-events-local
+azure_topic_location centralus-1
+
+azure_storage_account rmipactalocal
+azure_dest_portfolio_container parsedportfolios

cmd/dashboard/configs/test.conf

@@ -0,0 +1,9 @@
+env test
+min_log_level warn
+
+azure_event_topic pacta-events-test
+azure_topic_location westeurope-1
+
+azure_storage_account rmipactatest
+azure_dest_portfolio_container parsedportfolios
+