diff --git a/docs/adr/0036-member-list-query-paths.md b/docs/adr/0036-member-list-query-paths.md new file mode 100644 index 0000000000..0b9a8213ee --- /dev/null +++ b/docs/adr/0036-member-list-query-paths.md @@ -0,0 +1,46 @@ +# ADR-0036: Member list query picks one of three query shapes + +**Date**: 2025-11-19 +**Status**: accepted +**Deciders**: CDP team + +## Context + +The members list API (`queryMembersAdvanced` in `services/libs/data-access-layer/src/members/base.ts`) serves everything from a single-member lookup to the default "most active members of a project" page. It used to build one query shape for every request: start from `members`, join `memberSegmentsAgg`, enrichments and organizations, then sort and paginate. For large project groups that meant sorting millions of rows before applying `LIMIT`, and the planner could not pick a good plan for such different requests from one shape. Measured at the time, CNCF list requests took 45 to 50 seconds and Linux Foundation project requests regularly timed out. + +## Decision + +`buildQuery` in `services/libs/data-access-layer/src/members/queryBuilder.ts` inspects the filter and sort and emits one of three shapes. A filter that pins `m.id` to one value or an `IN` list of at most 100 IDs (with a segment) starts from `memberSegmentsAgg` for that segment; a request sorted by `activityCount` (the default) with no enrichment (`me.*`) or organization (`mo.*`) filters takes the top N rows of `memberSegmentsAgg` by activity count in a CTE, filtered inside the CTE, and joins back for the selected fields; everything else uses a fallback query from `members` that adds only the joins and CTEs the filter and fields need. The count runs as a separate query and is cached, so page navigation does not recount. + +## Alternatives Considered + +### One general query, tuned with indexes only + +- **Pros**: one code path; every filter and sort behaves the same way. +- **Cons**: the default page still sorts the whole segment's members before `LIMIT`, and the planner has to cover very different requests with one plan. +- **Why not**: this was the shape that timed out; indexes alone did not remove the full sort for the most common request. + +### Precompute the sorted member list (materialized view or similar) + +- **Pros**: the default page becomes a plain index read. +- **Cons**: another artifact to refresh and keep in sync with merges and new activity; it still has to support arbitrary filters and sorts. +- **Why not**: `memberSegmentsAgg` already holds per-segment activity counts, so an index on it (`idx_msa_segment_activitycount_desc_member` on `("segmentId", "activityCount" DESC, "memberId")`) gives the top-N read without a new pipeline. + +## Consequences + +### Positive + +- The default page reads a bounded slice of `memberSegmentsAgg` through the activity-count index instead of sorting every member. Measured at the time, CNCF requests dropped to about 11 seconds and Linux Foundation project requests to about 15 seconds without timeouts. +- Lookups of a few known members skip the top-N step and stay small. +- The fallback keeps every filter and sort working, joining enrichments and organizations only when the filter or selected fields reference them. + +### Negative + +- Three query shapes have to return the same rows for the same request, so a change to filtering or joins has to be made in each one. +- Path selection relies on string checks on the filter (`m.id = ...`, `m.id IN (...)`, the `me.` and `mo.` prefixes, other `m.*` fields), so it depends on the aliases the filter parser emits. +- The top-N CTE is capped at 50,000 candidates (10 times `limit + offset` when the filter uses other `m.*` fields). Requests whose page would pass the cap, such as large exports, fall back to the slower full query instead of returning a truncated page. + +### Risks + +- A new filter alias or a renamed table alias can silently route requests to the wrong path. The fallback still returns correct results, so the likely failure is slowness rather than wrong data, but there is no test that pins which path each request takes. +- The count is cached for up to six hours and depends on member write paths invalidating the member query cache; a write path that skips invalidation leaves a stale total next to fresh rows. diff --git a/docs/adr/README.md b/docs/adr/README.md index a9727f1ca7..b67c6e303b 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -42,6 +42,7 @@ Use the `/adr` skill in Claude Code to record new ADRs or query past decisions. | [ADR-0033](./0033-connector-emit-per-call-segment.md) | Connector emit() takes per-call segment overrides | accepted | 2026-09-30 | | [ADR-0034](./0034-connector-sync-deps-bag.md) | Connector syncs get database access through a typed per-connector deps bag | accepted | 2026-10-13 | | [ADR-0035](./0035-merge-suggestion-sources.md) | Track where merge suggestions come from | accepted | 2026-10-20 | +| [ADR-0036](./0036-member-list-query-paths.md) | Member list query picks one of three query shapes | accepted | 2025-11-19 | ## Why ADRs? diff --git a/services/apps/git_integration/README.md b/services/apps/git_integration/README.md index 3f366168f7..0b51f10be1 100644 --- a/services/apps/git_integration/README.md +++ b/services/apps/git_integration/README.md @@ -1,3 +1,340 @@ # Git Integration V2 -The Git integration is a Kubernetes-based service that processes Git repositories to extract contributor information from commits. It runs as worker pods that acquire repositories from the database queue, clone and process them, extract commits and maintainers, and send the processed data to Kafka for downstream ingestion. \ No newline at end of file +The Git integration is a Python worker service that processes Git repositories to extract +contributor information from commits. Each worker acquires one repository at a time from a +database-backed queue, clones it, runs a set of analysis services on the checkout, turns new +commits into activities, and hands those activities to the data sink through `integration.results` +and Kafka. + +For how its output reaches the rest of CDP, see +[Activity pipeline](../../../docs/design/activity-pipeline.md). This README covers the service's +own internals and how to run it. + +## Components + +The service is a FastAPI app (`src/crowdgit/server.py`) whose lifespan starts one +`RepositoryWorker` loop. `GET /` is a health check. On shutdown the worker stops picking new +repositories and gets up to `WORKER_SHUTDOWN_TIMEOUT_SEC` to finish the current one; after that the +run is cancelled. If the process is killed sooner, for example by a container grace period shorter +than that timeout, the repository keeps its lock and stays in `processing` (see Troubleshooting). + +| Component | Path | What it does | +| ----------------------------- | ---------------------------------------------- | ------------------------------------------------------------------------------------------------------------ | +| `RepositoryWorker` | `src/crowdgit/worker/repository_worker.py` | Polling loop: acquires a repository, runs the services, sets the final state, releases the lock | +| `CloneService` | `src/crowdgit/services/clone/` | Picks a full or incremental clone and yields clone batches | +| `SoftwareValueService` | `src/crowdgit/services/software_value/` | Runs the `software-value` Go binary ([README](src/crowdgit/services/software_value/README.md)) | +| `VulnerabilityScannerService` | `src/crowdgit/services/vulnerability_scanner/` | Runs the `vulnerability-scanner` Go binary ([README](src/crowdgit/services/vulnerability_scanner/README.md)) | +| `MaintainerService` | `src/crowdgit/services/maintainer/` | Finds and parses maintainer files, keeps `maintainersInternal` in sync | +| `AffiliationService` | `src/crowdgit/services/affiliation/` | Parses contributor-to-organization files kept in the repository | +| `LicenseService` | `src/crowdgit/services/license/` | Detects SPDX licenses with `licensee` | +| `CommitService` | `src/crowdgit/services/commit/` | Parses `git log` into activities | +| `QueueService` | `src/crowdgit/services/queue/` | Kafka producer for data sink messages | +| Database access | `src/crowdgit/database/crud.py` | All SQL the service runs | + +Maintainer and affiliation parsing call a Claude model on Amazon Bedrock +(`src/crowdgit/services/llm/bedrock.py`). + +## Processing a repository + +```mermaid +flowchart TD + A["Acquire repository"] --> B{"Fork?"} + B -- "yes" --> C["Validate parent"] + B -- "no" --> D["Clone"] + C --> D + D --> E["First batch: software value,
vulnerabilities, maintainers,
affiliations, licenses"] + E --> F{"Commit range complete?"} + F -- "no" --> G["Fetch deeper batch"] + G --> F + F -- "yes" --> H["Commits to activities"] + H --> I["Set state, release lock"] +``` + +### 1. Acquire + +The worker polls every `WORKER_POLLING_INTERVAL_SEC` and backs off `WORKER_ERROR_BACKOFF_SEC` +after a loop error. `acquire_repo_for_processing()` tries three queues in order and locks the +row it picks by setting `lockedAt` (`FOR UPDATE ... SKIP LOCKED`, so workers never share a +repository): + +1. **Onboarding**: `state = 'pending'`, ordered by `priority`, then `createdAt`. Skipped when + `integration.results` holds `MAX_INTEGRATION_RESULTS` rows or more (or the count query fails), + and when `MAX_CONCURRENT_ONBOARDINGS` repositories without a `lastProcessedCommit` are already + `processing`. +2. **Pending re-onboard** (Saturday and Sunday UTC only): `state = 'pending_reonboard'`. On + acquisition the worker clears `lastProcessedCommit` and `branch` and increments + `reOnboardingCount` in the same statement, so the repository goes through a fresh onboarding. +3. **Recurrent**: any state except `pending`, `processing`, `pending_reonboard` and + `auth_required`, last processed more than `REPOSITORY_UPDATE_INTERVAL_HOURS` ago (or + `FAILED_RETRY_INTERVAL_HOURS` for `failed`), ordered by `priority`, then oldest + `lastProcessedAt`. A hard-coded guard lets only one repository from a rate-limited Gerrit host + run at a time. + +Priorities are `0` urgent, `1` high and `2` normal. New rows default to `1`, and every finished +run resets the repository to `2`. Soft-deleted repositories (`deletedAt` set) are never acquired. + +### 2. Fork check + +When `forkedFrom` is set, the parent must exist with `state = 'completed'` and have been processed +within `REPOSITORY_UPDATE_INTERVAL_HOURS`. Otherwise the run ends with `requires_parent`. The +parent is used later to drop inherited maintainers and activities. + +### 3. Clone + +`CloneService.clone_batches_generator()` clones into a temporary directory: + +- **Full clone** (`--single-branch --no-tags`) when the repository has no `lastProcessedCommit` or + its remote default branch differs from the saved `branch`. The first batch is also the final one. +- **Incremental clone** otherwise: a `--depth=1` clone, then `git fetch --deepen` in batches that + start at 50 commits and double up to 10,000, until `lastProcessedCommit` is present and no + shallow boundary falls inside `lastProcessedCommit..HEAD`. Gerrit servers that fail on + `--deepen` get a single `--unshallow` fetch instead. + +Clone commands retry rate limits, network and remote 5xx errors three times with exponential +backoff. After each deepening fetch, a checkout larger than 10,000 MB is compacted with `git gc`. When the +first batch has been processed, the working tree is removed (only `.git` stays), and the temporary +directory is deleted at the end of the run. + +An incremental clone raises `ReOnboardingRequiredError`, and the run ends with +`pending_reonboard`, when either: + +- the clone is no longer shallow but `lastProcessedCommit` is missing (history was rewritten, for + example by a force push), or +- the time since `lockedAt` exceeds `STUCK_RECURRENT_REPO_TIMEOUT_HOURS`. The code also + defines `STUCK_ONBOARDING_REPO_TIMEOUT_HOURS` for repositories without a `lastProcessedCommit`, + but those always take the full-clone path, which returns before the timeout check, so it never + applies. + +### 4. First-batch services + +These need the working tree, so they run once, on the first batch, in this order. A failure in one +does not stop the run. Each records its result in `git."serviceExecutions"` except licenses, which +have no operation type there and only write to the logs. + +- **Software value** and **vulnerability scanner** run their Go binaries on the checkout. The + binaries write their results to the database themselves; see their READMEs. +- **Maintainers** run when the interval has elapsed since `lastMaintainerRunAt`: + `MAINTAINER_UPDATE_INTERVAL_HOURS` if a maintainer file was found last time, + `MAINTAINER_RETRY_INTERVAL_DAYS` if not. Detection order: + 1. A CNCF `.project` repository uses its `maintainers.yaml` (or `.yml`) directly, without the + LLM. + 2. The file saved in `maintainerFile`, if it still exists. + 3. Candidate files found by name (`MAINTAINERS`, `CODEOWNERS`, `OWNERS`, `GOVERNANCE.md` and + similar), scored, then filtered by an LLM classifier. Root-level candidates are parsed first + and their results combined; otherwise the top subdirectory candidate is tried. + 4. As a last resort, the LLM picks a file from the repository's file list. + + Maintainers resolve to verified `memberIdentities` by GitHub username, then by email. The + service upserts new `(identity, role)` pairs into `maintainersInternal` and sets `endDate` on + pairs that disappeared from the file. The first run for a repository (no `lastMaintainerRunAt`) + only inserts; end-dating starts from the second run. A pair whose GitHub username or email is still in the file + but no longer resolves to a verified identity keeps its open row; it is only end-dated once it + leaves the file or its identity is deleted. For forks, maintainers already listed on the parent are dropped. In a + CNCF project that has a `.project` repository, the other repositories skip maintainer extraction, + and the `.project` file end-dates their maintainer rows. + +- **Affiliations** run every `AFFILIATION_UPDATE_INTERVAL_HOURS` once a file is known, or every + `AFFILIATION_RETRY_INTERVAL_DAYS` while none is, and track the file, its hash and a parsed snapshot in + `git."repoAffiliationRegistry"`. What the parsed rows mean for affiliation is described in + [How CDP affiliates activities with organizations](../../../docs/knowledge/flow-affiliation.md). +- **Licenses** run `licensee detect` and write SPDX identifiers with confidence of at least 98 to + `public.repositories.licenses`. A license file that matches nothing reliably is stored as + `NOASSERTION`; no license file stores an empty list. A missing binary or a timeout is logged as a warning and stores an empty list. Any other + non-zero `licensee` exit is logged at info level as "found no license", so it looks exactly like + a repository without a license file, in the data and in the logs. + +### 5. Commits + +On the final batch, `CommitService` runs `git log --numstat` over the whole default branch (full +clone) or `lastProcessedCommit..HEAD` (incremental clone). It splits the output into chunks of 250 +commits, processes two chunks at a time, and skips commits whose hash, dates or emails fail +validation. Each valid commit produces: + +- `authored-commit` for the author, with the commit hash as `sourceId`; +- `committed-commit` for the committer, when author and committer differ; +- one activity per recognized message trailer (`Signed-off-by`, `Reviewed-by`, `Tested-by`, + `Co-authored-by` and others), mapped through `activitymap.py` and named like + `signed-off-commit`. + +The activity body is the commit message, capped at 10,000 characters. When `reOnboardingCount` is +above zero, activities carry `attributes.cycle = 'onboarding-'`, which lets +downstream queries tell re-onboarding cycles apart. For forks, activities that already exist on +the parent (same timestamp, type and `sourceId` in `activityRelations`) are dropped. + +Each chunk is written to `integration.results` and, at the same time, published to +`CROWD_KAFKA_TOPIC` as a `process_integration_result` message for the data sink. After all chunks +succeed, the worker saves the latest commit as `lastProcessedCommit` along with the default +branch. + +### 6. Finish + +The worker sets the final state, `lastProcessedAt` and `priority = 2`, then clears `lockedAt`. + +| State | Set when | +| ------------------- | ---------------------------------------------------------------------------------------- | +| `pending` | Row created, or reset for re-onboarding; waits in the onboarding queue | +| `processing` | A worker holds the repository | +| `completed` | The run finished | +| `failed` | Any other error; retried after `FAILED_RETRY_INTERVAL_HOURS` | +| `requires_parent` | Fork whose parent is missing or not recently completed; retried on the recurrent cadence | +| `pending_reonboard` | History rewrite or timeout; re-onboarded on the next weekend | +| `auth_required` | Clone failed with an authentication or "not found" error; never picked again | + +## Tables + +| Table | Use | +| ------------------------------- | ----------------------------------------------------------------------------------------------------------- | +| `public.repositories` | Repository URL, segment, `gitIntegrationId`, `forkedFrom`, `deletedAt`; the service writes `licenses` | +| `git."repositoryProcessing"` | One row per repository: state, priority, lock, last commit and branch, maintainer file, re-onboarding count | +| `git."serviceExecutions"` | One row per service run: operation, status, error code and message, duration, metrics | +| `git."repoAffiliationRegistry"` | Affiliation file, hash, parsed snapshot and last run per repository | +| `maintainersInternal` | Maintainers per repository and role, with start and end dates | +| `integration.results` | Activities waiting for the data sink | + +Repository rows are created outside this service: `services/libs/data-access-layer/src/repositories/` +inserts into `public.repositories` and `git."repositoryProcessing"` together, and +`restoreRepositories()` resets the processing row to `pending` when a soft-deleted or changed +repository is restored. + +`git."serviceExecutions"` keeps about 14 days: an insert trigger deletes older rows on a random 1% +of inserts. `operationType` is one of `Clone`, `Commit`, `Maintainer`, `Affiliation`, +`SoftwareValue` or `VulnerabilityScanner`, and `errorCode` values are listed in +`src/crowdgit/enums.py`. + +## Configuration + +All settings are environment variables read in `src/crowdgit/settings.py`. + +| Variable | Default | Purpose | +| ----------------------------------------------------------------------------------------------------- | --------- | -------------------------------------------------------------------------------------- | +| `CROWD_DB_WRITE_HOST`, `CROWD_DB_PORT`, `CROWD_DB_USERNAME`, `CROWD_DB_PASSWORD`, `CROWD_DB_DATABASE` | required | Postgres connection | +| `CROWD_KAFKA_BROKERS`, `CROWD_KAFKA_TOPIC` | required | Kafka producer target | +| `CROWD_KAFKA_EXTRA` | unset | JSON with `ssl` and `sasl` settings, in the kafkajs format | +| `CROWD_AWS_BEDROCK_ACCESS_KEY_ID`, `CROWD_AWS_BEDROCK_SECRET_ACCESS_KEY`, `CROWD_AWS_BEDROCK_REGION` | unset | Bedrock client for maintainer and affiliation parsing | +| `CROWD_SSO_LF_TENANT_ID` | built in | Tenant ID written on results | +| `WORKER_POLLING_INTERVAL_SEC` | `5` | Sleep between loop iterations | +| `WORKER_ERROR_BACKOFF_SEC` | `10` | Sleep after a loop error | +| `WORKER_SHUTDOWN_TIMEOUT_SEC` | `3600` | Time to finish the current repository on shutdown | +| `REPOSITORY_UPDATE_INTERVAL_HOURS` | `24` | Recurrent processing interval; also the fork parent freshness window | +| `FAILED_RETRY_INTERVAL_HOURS` | `6` | Retry interval for `failed` repositories | +| `MAX_CONCURRENT_ONBOARDINGS` | `3` | Onboarding repositories processed at once across all workers | +| `MAX_INTEGRATION_RESULTS` | `5000000` | `integration.results` size above which onboarding pauses | +| `STUCK_RECURRENT_REPO_TIMEOUT_HOURS` | `4` | Incremental clone timeout before `pending_reonboard` | +| `STUCK_ONBOARDING_REPO_TIMEOUT_HOURS` | `12` | Meant for repositories without `lastProcessedCommit`; never applies today (see step 3) | +| `MAINTAINER_UPDATE_INTERVAL_HOURS` | `24` | Maintainer refresh when a file was found | +| `MAINTAINER_RETRY_INTERVAL_DAYS` | `30` | Maintainer retry when no file was found | +| `AFFILIATION_UPDATE_INTERVAL_HOURS` | `24` | Affiliation refresh when a file was found | +| `AFFILIATION_RETRY_INTERVAL_DAYS` | `30` | Affiliation retry when no file was found | + +`CROWD_KAFKA_USER` and `CROWD_KAFKA_PASSWORD` are read but not used; credentials come from +`CROWD_KAFKA_EXTRA`. + +## Running locally + +The service needs `git`, `ripgrep`, `licensee` and the two Go binaries, so it runs in Docker +(`scripts/services/docker/Dockerfile.git_integration` builds all of them). + +1. Start the local infrastructure (`./cli scaffold up` in `scripts/`), which creates the + `crowd-bridge` network, Postgres and Kafka. +2. `make setup` installs `uv` and the dev dependencies for linting and tests. +3. `make run` starts `git-integration-dev` from `scripts/services/git-integration.yaml`. It reads + the `backend/.env.*` files, mounts `src/` and reloads on change. The health check is on port 8085. +4. To give the worker something to do, connect a Git, GitHub, GitLab or Gerrit integration, or + insert a row into `public.repositories` and a matching `git."repositoryProcessing"` row. + +Other targets: `make lint`, `make format`, `make rebuild`, `make cmd -- `, and +`make add_package name=`. `make test` runs pytest locally on commit and activity extraction +against a test repository built from a seed file, with the queue service mocked. It takes optional +`test=`, `repo=` and `expected=` arguments; see +[src/test/README.md](src/test/README.md). + +## Troubleshooting + +Start with the repository's processing row and its recent service runs: + +```sql +SELECT r.id, r.url, rp.state, rp.priority, rp."lockedAt", rp."lastProcessedAt", + rp."lastProcessedCommit", rp.branch, rp."maintainerFile", rp."reOnboardingCount" +FROM public.repositories r +JOIN git."repositoryProcessing" rp ON rp."repositoryId" = r.id +WHERE r.url = 'https://github.com//' AND r."deletedAt" IS NULL; + +SELECT "operationType", status, "errorCode", "errorMessage", "executionTimeSec", metrics, "createdAt" +FROM git."serviceExecutions" +WHERE "repoId" = '' +ORDER BY "createdAt" DESC +LIMIT 20; +``` + +**A repository stays in `processing`.** The worker clears `lockedAt` in a `finally` block, so a +lock that outlives any reasonable run means the worker process died (for example, out of memory or +killed during shutdown). Neither queue picks a locked row, and a dead onboarding run still counts +against `MAX_CONCURRENT_ONBOARDINGS`. Check that no worker is running it, then release it. A +repository that was never processed goes back to `pending`, because the recurrent queue needs a +`lastProcessedAt` and would never pick it up as `failed`: + +```sql +UPDATE git."repositoryProcessing" rp +SET state = CASE WHEN rp."lastProcessedAt" IS NULL THEN 'pending' ELSE 'failed' END, + "lockedAt" = NULL, "updatedAt" = NOW() +FROM public.repositories r +WHERE rp."repositoryId" = r.id AND r.url = 'https://github.com//'; +``` + +**Re-onboard a repository.** Reset the processing row so the next run does a full clone and +reprocesses every service except affiliations, which keep their own interval in +`git."repoAffiliationRegistry"` (`AFFILIATION_UPDATE_INTERVAL_HOURS` after a file was found, +`AFFILIATION_RETRY_INTERVAL_DAYS` otherwise). Set `priority = 0` as well to put it at the front of the onboarding +queue. Skip rows in `processing`. + +```sql +UPDATE git."repositoryProcessing" rp +SET "lastProcessedAt" = NULL, "lastProcessedCommit" = NULL, + "lastMaintainerRunAt" = CASE WHEN rp."lastMaintainerRunAt" IS NULL THEN NULL + ELSE TIMESTAMPTZ '1970-01-01' END, + branch = NULL, "lockedAt" = NULL, state = 'pending', "updatedAt" = NOW() +FROM public.repositories r +WHERE rp."repositoryId" = r.id + AND r.url = 'https://github.com//' + AND rp.state != 'processing'; +``` + +`lastMaintainerRunAt` is set to an old date rather than cleared: with no previous run the service +takes its first-run path, which only inserts maintainers and never end-dates removed ones. + +To force the affiliation check in the same run, also clear its last run: + +```sql +UPDATE git."repoAffiliationRegistry" ar +SET "lastRunAt" = NULL, "updatedAt" = NOW() +FROM public.repositories r +WHERE ar."repoId" = r.id AND r.url = 'https://github.com//'; +``` + +A manual re-onboard does not change `reOnboardingCount`, so its activities keep the current +cycle: no `cycle` attribute when the count is 0, otherwise the same `onboarding-` as the last +re-onboard, which mixes them with that cycle's activities. When the reason is rewritten history, +increment `reOnboardingCount` in the same update to mark a new cycle. + +**Many repositories in `pending_reonboard`.** These wait for the weekend by design. To run one +sooner, re-onboard it manually as above. + +**Onboarding is slow or paused.** Check the `integration.results` row count against +`MAX_INTEGRATION_RESULTS` and the number of `processing` rows without a `lastProcessedCommit` +against `MAX_CONCURRENT_ONBOARDINGS`. Raising `MAX_CONCURRENT_ONBOARDINGS` or adding workers +increases onboarding throughput; each worker processes one repository at a time. + +**A fork stays in `requires_parent`.** Its parent (`forkedFrom` matched against +`public.repositories.url`) is missing, soft-deleted, not `completed`, or older than +`REPOSITORY_UPDATE_INTERVAL_HOURS`. The fork is retried on the recurrent cadence and succeeds once +the parent has a recent completed run. + +**A repository is `auth_required`.** The clone failed with an authentication, 401 or "Repository +not found" error. No queue picks it again; once the repository is reachable, re-onboard it. + +**Maintainers are missing.** Look at the `Maintainer` rows in `git."serviceExecutions"`. +`no-maintainer-file` waits `MAINTAINER_RETRY_INTERVAL_DAYS` before the next attempt, +`maintainer-interval-not-elapsed` is the normal skip between runs, and +`maintainer-skipped-project-level-source` means a CNCF `.project` repository owns the project's +maintainers. Extracted maintainers without a verified GitHub or email identity in CDP are logged +and skipped; if they had a row before, it stays open rather than being end-dated. diff --git a/services/apps/git_integration/src/crowdgit/services/software_value/README.md b/services/apps/git_integration/src/crowdgit/services/software_value/README.md index beb2f5608d..2c98047dbd 100644 --- a/services/apps/git_integration/src/crowdgit/services/software_value/README.md +++ b/services/apps/git_integration/src/crowdgit/services/software_value/README.md @@ -1,115 +1,105 @@ -# Project Overview +# Software value -This tool analyzes source code repositories and saves some data about the to a database, including the estimated cost -to create the project, and some statistics about the programming languages used. +This tool analyzes a source code repository and saves data about it to the Insights database: the +estimated cost to develop the project, and statistics about the programming languages it uses. -It leverages a tool called [scc](https://github.com/boyter/scc) to gather the statistics about the codebase. +It is a Go binary that wraps [scc](https://github.com/boyter/scc). The git integration worker +calls it as a subprocess. -It was a simple proof of concept that graduated into a real tool, so please feel free to contribute and improve it. +## How it runs -## Getting started +`software_value_service.py` runs the binary at `/usr/local/bin/software-value` once per repository +processing run, on the first clone batch (`is_first_batch` in `crowdgit/worker/repository_worker.py`). That +batch is a full clone for a new repository or a changed default branch, and a depth-1 clone otherwise; +either way the working tree is the head of the default branch, which is what scc measures. -This explains how to set up and run the software-value tool locally. -To run the tool on a production server, follow the Ansible instructions below. +The worker picks up a repository again when its last processing is older than +`REPOSITORY_UPDATE_INTERVAL_HOURS` (default 24, in `crowdgit/settings.py`), or +`FAILED_RETRY_INTERVAL_HOURS` after a failed run, so the numbers refresh at most at that rate. They +reach Insights only after the Tinybird copy pipes that read them next run. -### Requirements +Before it calls the binary, the service: -- Go (latest stable version) -- PostgreSQL 14+ -- [scc](https://github.com/boyter/scc) binary (path set in config) +- skips repositories listed in `_SOFTWARE_VALUE_EXCLUDED_REPO_IDS`; +- measures the clone with `du -sb` and, at 10 GB or more, passes `--no-large`, which makes scc skip + files larger than 100 MB to avoid running out of memory. -The version of scc tested at this time was v3.5.0. -Using a different version may break this tool because the output format may have changed. +The binary always exits with code 0 and prints a JSON status on stdout: -### Running the tool locally +```json +{ "status": "failure", "error_code": "NO_COST_FOUND", "error_message": "..." } +``` -1. Clone the repository and navigate to the project directory. +The service records the status, error code and duration as a service execution with operation type +`SOFTWARE_VALUE`. Error codes are defined in `types.go`. -2. Install dependencies: - `go mod download` +## What it computes -3. Create your configuration file: - - Copy `config.toml.example` to `config.toml`. - - Edit `config.toml` to set your database credentials, paths, and other settings. +For the repository at the path given as the only argument, the binary: -4. Place the scc binary in the correct location: - - Ensure the `scc` binary is in the path specified in your `config.toml` file. +1. Reads the `origin` remote with `git remote get-url origin` and rewrites `git@`, `ssh://git@` + and `http://` URLs to `https://`. This URL is the key in both tables. +2. Runs `scc --format=short` and parses the COCOMO line "Estimated Cost to Develop" into a + dollar amount. A cost of zero fails the run with `NO_COST_FOUND` and saves nothing. +3. Runs `scc --format=json` for per-language counts: bytes, code bytes, lines, code, comment and + blank lines, complexity, file count and weighted complexity. +4. Upserts the cost into `project_costs` and one row per language into `language_stats`. -5. Build and run the application: `go run ./` +## Data model +The tables live in the Insights database, not the CDP database. `db.sql` defines them: +| Table | Key | Holds | +| ---------------- | --------------------------- | --------------------------------------- | +| `project_costs` | `repo_url` | `estimated_cost` (whole dollars) | +| `language_stats` | `(repo_url, language_name)` | scc's per-language counts for that repo | -## Deploying with Ansible +`language_stats.repo_url` references `project_costs` with `ON DELETE CASCADE`. -In the `ansible` directory, there is an Ansible playbook that can be used to deploy the tool automatically. -You can run it with the following command: +## Where the numbers go -```bash -ansible-playbook -i inventory.ini playbook.yml -``` +`project_costs` is replicated to Tinybird as the datasource +`services/libs/tinybird/datasources/softwareValueProjectCosts.datasource`. From there: -For this, you will need an `ansible/inventory.ini` file that specifies the target hosts and their SSH access details, -as well as an `ansible/group_vars/software_value.yaml` with the necessary variables for the config.toml template. +- `repositories_populated_copy.pipe` exposes the cost per repository as `softwareValue`; +- `segments_aggregates_with_ids_copy.pipe` sums the cost of a project's repositories into the + project's `softwareValue`. -Here's an example of what your `inventory.ini` file might look like: - -```ini -[software_value] -100.10.20.30 ansible_user=ubuntu -``` +`language_stats` has no Tinybird datasource in this repository. -And these are the group variables that you need, with the variables replaced for your environment: - -```yaml -db_insights_host: "10.10.10.10" -db_insights_port: 5432 -db_insights_user: "lfx" -db_insights_password: "mightypassword" -db_insights_dbname: "lfx_insights_software_value" -db_insights_sslmode: "require" -db_insights_pool_max: 10 -db_insights_readonly: false - -db_cm_host: "10.20.30.40" -db_cm_port: 5432 -db_cm_user: "cm" -db_cm_password: "passwordmight" -db_cm_dbname: "cm" -db_cm_sslmode: "require" -db_cm_pool_max: 10 -db_cm_readonly: true - -scc_path: "{{ svc_home }}/scc" -target_path: "/path/to/repositories" -batch_size: 100 -``` +## Configuration -If you modified the code and want to build an updated binary to be uploaded to the server, you can use: +The binary reads the Insights database connection from environment variables: -```bash -go build -ldflags "-w" -o software-value ./ && strip software-value -```` +| Variable | Use | +| ------------------------ | ------------------------------- | +| `INSIGHTS_DB_WRITE_HOST` | Host | +| `INSIGHTS_DB_PORT` | Port | +| `INSIGHTS_DB_USERNAME` | User | +| `INSIGHTS_DB_PASSWORD` | Password | +| `INSIGHTS_DB_DATABASE` | Database name | +| `INSIGHTS_DB_SSLMODE` | `sslmode` for the connection | +| `INSIGHTS_DB_POOL_MAX` | Maximum connections in the pool | +| `INSIGHTS_DB_READONLY` | `true` opens read-only sessions | -## Running automatically +It finds `scc` on `PATH`. -You should not need to do this. -These instructions are provided for just for completeness. -You should use the Ansible playbook to deploy the tool instead. +## Build -There are Systemd service and timer unit templates included in the `ansible/templates` directory. -The templates are named `lfx-software-value.service.j2` and `lfx-software-value.timer.j2`. -Make copies, rename them to `lfx-software-value.service` and `lfx-software-value.timer`, and edit them to replace the -variables with the correct values for your situation. -Finally, copy them to`/etc/systemd/system/` in the machine you want to run the tool, -and enable them with the following commands: +`scripts/services/docker/Dockerfile.git_integration` builds the binary in a Go stage, installs scc +v3.5.0 with `go install`, and copies both to `/usr/local/bin` in the git integration image. The +parser depends on scc's output format, so test before changing the scc version. -```bash -sudo systemctl daemon-reload -sudo systemctl enable lfx-software-value.timer -``` +## Running locally -Then start the timer with: +Requirements: Go (the version in `go.mod`), PostgreSQL with the tables from `db.sql`, and scc +v3.5.0 on `PATH`. ```bash -sudo systemctl start lfx-software-value.timer +go mod download +export INSIGHTS_DB_WRITE_HOST=localhost INSIGHTS_DB_PORT=5432 INSIGHTS_DB_USERNAME=... \ + INSIGHTS_DB_PASSWORD=... INSIGHTS_DB_DATABASE=... INSIGHTS_DB_SSLMODE=disable INSIGHTS_DB_POOL_MAX=4 +go run ./ [--no-large] /path/to/cloned/repository ``` + +The repository must be a git clone with an `origin` remote. diff --git a/services/apps/snowflake_connectors/README.md b/services/apps/snowflake_connectors/README.md new file mode 100644 index 0000000000..d4d448f008 --- /dev/null +++ b/services/apps/snowflake_connectors/README.md @@ -0,0 +1,184 @@ +# Snowflake connectors worker + +`snowflake-connectors-worker` exports source data from Snowflake to S3 as Parquet, transforms each row into CDP +activities, and sends them to `data_sink_worker`. It runs as a Temporal worker on the `snowflakeConnectors` task queue +and also runs a long-lived consumer loop in the same process. + +For where this service sits in the ingestion pipeline, see +[`docs/design/activity-pipeline.md`](../../../docs/design/activity-pipeline.md). Shared Snowflake, S3 and job-table code +lives in [`services/libs/snowflake`](../../libs/snowflake). + +## Components + +| Component | Code | What it does | +| -------------------- | ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------ | +| Export workflow | `src/workflows/exportWorkflow.ts`, `src/activities/exportActivity.ts` | Runs `COPY INTO` in Snowflake to write Parquet files to S3, one child workflow per platform and source | +| Transformer consumer | `src/consumer/transformerConsumer.ts` | Claims pending export jobs, reads the Parquet files, transforms rows and emits activities | +| Cleanup workflow | `src/workflows/cleanupWorkflow.ts`, `src/activities/cleanupActivity.ts` | Deletes S3 files for jobs that finished cleanly | +| Integration resolver | `src/core/integrationResolver.ts` | Maps a row's project to a CDP segment and integration | +| Platform registry | `src/integrations/index.ts`, `src/integrations/types.ts` | Lists each platform's data sources, with a query builder and transformer per source | + +`src/index.ts` starts the worker: it registers both Temporal schedules, starts the consumer, and then starts the +Temporal worker. On `SIGINT` or `SIGTERM` the consumer stops claiming jobs and tries to finish the current one; the process +force-exits after 2 hours, or sooner if the container is killed. An interrupted job keeps `processingStartedAt` with no +`completedAt` or `error`: no consumer claims it again, and the activities it already emitted stay. Re-run it by clearing +`processingStartedAt`. + +## Sources + +A platform has one or more data sources. Each source maps to a Snowflake table, or a joined set of tables, and has its +own `buildSourceQuery.ts` and `transformer.ts` under `src/integrations///`. Each source is exported, +tracked and transformed on its own. + +| Platform | Source | Activity types | +| ------------ | --------------------- | ---------------------------------------------- | +| `cvent` | `event-registrations` | `registered-event`, `attended-event` | +| `tnc` | `enrollments` | `enrolled-certification`, `enrolled-training` | +| `tnc` | `certificates` | `issued-certification` | +| `tnc` | `courses` | `attempted-course`, `attempted-exam` | +| `meetings` | `meeting-attendance` | `invited-meeting`, `attended-meeting` | +| `committees` | `committees` | `added-to-committee`, `removed-from-committee` | + +A platform runs only when it is listed in `CROWD_SNOWFLAKE_ENABLED_PLATFORMS`. Code for a platform that is not listed +is never exported or transformed. + +To add a platform or a source, use the +[`scaffold-snowflake-connector`](../../../.claude/skills/scaffold-snowflake-connector/SKILL.md) skill. It lists every +touch point, including the `PlatformType` and activity type enums, the activity types migration and the registry entry. + +## How it works + +### 1. Export + +The `snowflake-s3-export` schedule starts `snowflakeS3ExportScheduler` every day at 00:20. For each enabled platform it +starts one `exportSnowflakeData` child workflow per source. + +Each child runs `executeExport(platform, sourceName)`: + +1. Reads the latest `exportStartedAt` among this source's jobs that completed without an error. With no such job, the + export is a full export. +2. Calls the source's `buildSourceQuery(sinceTimestamp)`. With a timestamp, the query returns rows updated since then, + plus every row for projects whose segment was created since then. The TNC sources filter on event time instead + (enrollment, certificate issue or course action), so a correction to an older TNC record in an existing project is + not exported again. +3. Runs `COPY INTO` in batches of 10,000 rows. Each batch writes one Parquet file under + `/////
/`. +4. Inserts one row per file into `integration."snowflakeExportJobs"`. + +Source queries keep only rows whose project matches a CDP segment. Outside production (`NODE_ENV` is not +`production`), each query also limits itself to a single project. + +### 2. Transform + +The consumer loop claims the oldest unclaimed job with `FOR UPDATE SKIP LOCKED`, so several consumers never take the +same job. For each job it: + +1. Looks up the source's transformer from the job's `platform` and `sourceName`. +2. Streams the Parquet rows from S3 and calls `safeTransformRow()` on each. A transformer returns one activity, several + activities, or `null` to skip the row. A transformer that throws is logged and counted as a skip. +3. Resolves each activity's segment and integration, then sends the activity to `data_sink_worker` in the queue message + through `DataSinkWorkerEmitter`. +4. Marks the job completed with its metrics, or failed with the error. + +When no job is pending, the loop sleeps for 10 seconds and doubles the wait on each empty poll, up to 30 minutes. A +claimed job resets the wait. + +### 3. Resolve segment and integration + +Every transformed row carries a segment reference built from the row's `PROJECT_SLUG` and `PROJECT_ID` columns. The +resolver matches it, case-insensitively, against a `subproject` segment's `slug` and `sourceId`. Rows with no matching +segment are skipped. + +If the segment has no integration for the platform, the resolver creates one with status `done`. A Redis lock stops two +consumers from creating the same integration. Results are cached in Redis for 24 hours, and the cache is warmed at +startup for all enabled platforms. + +### 4. Clean up + +The `snowflake-s3-cleanup` schedule starts `snowflakeS3CleanupScheduler` every day at 02:00. It deletes the S3 file and +sets `cleanedAt` for each job that completed more than 24 hours ago with no error and zero skipped rows. Files for +failed jobs and jobs with skipped rows stay in S3 for inspection. + +### Shared rules in transformers + +`src/core/transformerBase.ts` and `src/core/personalEmail.ts` hold rules that every transformer uses: + +- `cleanIdentityValue()` treats empty strings and placeholders such as `none`, `null`, `n/a` and `-` as missing. +- `buildMemberIdentities()` adds a verified email identity, plus username identities from the platform username, the + LFID username or, when neither exists, the email. +- `cleanPersonalEmail()` drops known shared meeting-account emails, so they never become member identities. + +## Tables + +`integration."snowflakeExportJobs"` holds one row per exported Parquet file. + +| Column | Meaning | +| -------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `platform`, `"sourceName"` | Which source produced the file | +| `s3_path` | Parquet file location (unique) | +| `"exportStartedAt"` | When the export run started; the next incremental export starts here | +| `"processingStartedAt"` | Set when a consumer claims the job | +| `"completedAt"` | Set when processing finishes, whether it succeeded or failed | +| `error` | Error message for a failed job | +| `"cleanedAt"` | When the S3 file was deleted | +| `metrics` | `exportedRows` and `exportedBytes` from the export, then `transformedCount`, `skippedCount`, `transformSkippedCount`, `resolveSkippedCount` and `processingDurationMs` | + +The table is created in `V1771344764__addSnowflakeExportTable.sql` and `V1772548920__addSourceNameToSnowflakeExportJobs.sql` +under `backend/src/database/migrations/`. + +The resolver also reads `segments` and reads and writes `integrations`. + +## Configuration + +| Variable | Purpose | +| -------------------------------------- | -------------------------------------------------------------------------------- | +| `CROWD_SNOWFLAKE_ENABLED_PLATFORMS` | Comma-separated platforms to run. Empty means nothing is exported or transformed | +| `CROWD_SNOWFLAKE_ACCOUNT` | Snowflake account | +| `CROWD_SNOWFLAKE_USERNAME` | Snowflake user | +| `CROWD_SNOWFLAKE_TOKEN` | Programmatic access token. When set, it is used instead of the private key | +| `CROWD_SNOWFLAKE_PRIVATE_KEY` | Private key for key-pair authentication | +| `CROWD_SNOWFLAKE_DATABASE` | Snowflake database | +| `CROWD_SNOWFLAKE_WAREHOUSE` | Snowflake warehouse | +| `CROWD_SNOWFLAKE_ROLE` | Snowflake role | +| `CROWD_SNOWFLAKE_STORAGE_INTEGRATION` | Snowflake storage integration that `COPY INTO` writes through | +| `CROWD_SNOWFLAKE_S3_BUCKET_PATH` | S3 path prefix for exported files | +| `CROWD_SNOWFLAKE_S3_REGION` | S3 region | +| `CROWD_SNOWFLAKE_S3_ACCESS_KEY_ID` | S3 credentials used to read and delete files | +| `CROWD_SNOWFLAKE_S3_SECRET_ACCESS_KEY` | S3 credentials used to read and delete files | + +The worker also uses the standard CDP Postgres, Redis, Kafka queue and Temporal settings. The four +`CROWD_SNOWFLAKE_S3_*` variables and `CROWD_SNOWFLAKE_STORAGE_INTEGRATION` are required at startup. + +## Running locally + +The `dev:local` script loads `backend/.env.dist.local` and `backend/.env.override.local`, then runs the worker with +reload on changes in `src/` and `services/libs/`: + +```bash +pnpm run dev:local +``` + +To run an export or a cleanup outside the schedule: + +```bash +pnpm run trigger-export +pnpm run trigger-cleanup +``` + +Both scripts start the workflow on the `snowflakeConnectors` task queue, so a worker must be running to pick it up. + +To run the worker in Docker with the rest of the local stack, use `./cli service snowflake-connectors-worker up`. + +## Troubleshooting + +- **Nothing is exported.** Check that the platform is in `CROWD_SNOWFLAKE_ENABLED_PLATFORMS`. The schedules are + created once; if `snowflake-s3-export` already exists in Temporal, a change to its schedule in code does not apply + until the schedule is updated or deleted there. +- **Activities are missing for a project.** Look at `resolveSkippedCount` in the job's `metrics`. A non-zero count means + the resolver returned nothing for some rows: either no `subproject` segment matches their `PROJECT_SLUG` and + `PROJECT_ID` (`Segment not found, skipping`), or the integration could not be found after waiting on another + consumer's creation lock (`Integration not found after lock contention, skipping`). The log line tells them apart. +- **Rows are skipped by the transformer.** Look at `transformSkippedCount` and the `Failed to transform row, skipping` + log lines, which list the row's keys. +- **A job failed.** Its `error` column holds the message. The S3 file stays in place, but the job is not claimed again. +- **A new source got a full export.** That is expected: a source with no completed job starts without a timestamp. diff --git a/services/cronjobs/archived_repositories/README.md b/services/cronjobs/archived_repositories/README.md index ae93c0aa8d..b65b616b8e 100644 --- a/services/cronjobs/archived_repositories/README.md +++ b/services/cronjobs/archived_repositories/README.md @@ -21,8 +21,59 @@ It has two main processes: involves calling the GitHub and GitLab APIs to check if the repository is archived, and updating the repositories in the database. +## How it works + +### Selecting repositories + +Each run of the main process reads up to `ARCHIVED_REPOSITORIES_CHECKER_BATCH_SIZE` (default 10,000) URLs from +`public.repositories`. It picks rows that are not soft-deleted, whose URL starts with `https://github.com/` or +`https://gitlab.com/`, and that were never checked or were last checked more than 3 days ago. Rows never checked come +first, then the ones checked longest ago. + +It enqueues each URL into one of two BullMQ queues, `github-archived-repo-checker` or `gitlab-archived-repo-checker`. +A failed job is retried up to 3 times with exponential backoff (4 attempts in total). + +### Rate limits + +Each queue has its own worker, with a concurrency of 5 and a limiter set below the provider's published limit: + +| Queue | Limiter | Provider limit | +| ------ | --------------------- | --------------------------------------- | +| GitHub | 3,500 jobs per hour | 5,000 requests per hour (authenticated) | +| GitLab | 1,500 jobs per minute | 2,000 requests per minute | + +The limits are set in `src/workers.ts`, not through environment variables. + +### Marking archived and excluded + +The worker calls the GitHub or GitLab project API and sets two flags: + +- `archived` takes the provider's `archived` value, and `excluded` is `false`. +- A 403 or 404 response means the repository is gone or not accessible, so the worker sets `excluded` to `true` and + `archived` to `false`. +- Repositories named `.github` on GitHub are always set to `excluded`, whatever their archive status. +- Any other error fails the job, and BullMQ retries it. + +Each processed repository gets its `archived`, `excluded`, `lastArchivedCheckAt` and `updatedAt` columns updated in +`public.repositories`. The table is created in +[`V1766661879__createRepositoriesTable.sql`](../../../backend/src/database/migrations/V1766661879__createRepositoriesTable.sql). +Sequin replicates it to the Tinybird `repositories` datasource, where health score and security pipes filter on these +flags. [`docs/design/activity-pipeline.md`](../../../docs/design/activity-pipeline.md) shows how Sequin replication +reaches Tinybird. + ## Configuration +| Variable | Purpose | +| --------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------ | +| `ARCHIVED_REPOSITORIES_CHECKER_GITHUB_TOKEN` | GitHub API token | +| `ARCHIVED_REPOSITORIES_CHECKER_GITLAB_TOKEN` | GitLab API token | +| `CROWD_DB_WRITE_HOST`, `CROWD_DB_READ_HOST`, `CROWD_DB_PORT`, `CROWD_DB_USERNAME`, `CROWD_DB_PASSWORD`, `CROWD_DB_DATABASE` | PostgreSQL connection. Queries and updates both use the write host | +| `CROWD_REDIS_HOST`, `CROWD_REDIS_PORT`, `CROWD_REDIS_USERNAME`, `CROWD_REDIS_PASSWORD` | Redis connection for BullMQ | +| `ARCHIVED_REPOSITORIES_CHECKER_REDIS_DB` | Redis database index | +| `ARCHIVED_REPOSITORIES_CHECKER_BATCH_SIZE` | Optional. URLs enqueued per run (default 10,000) | + +All variables except the batch size are required; the process exits at startup if one is missing. + ### For production In production in our Kubernetes cluster, the environment variables can be set in a ConfigMap which is loaded into the @@ -37,6 +88,10 @@ You just need to copy the file `.env.example` to `.env` and fill in the required You need access to a PostgreSQL database with the necessary repositories table, a Redis instance for the tasks queue, and the GitHub and GitLab API tokens. +`docker-compose.yaml` starts a local PostgreSQL and Redis along with the main and workers containers, which read +`.env` through a volume. The local PostgreSQL starts empty, so create the `public.repositories` table before running +the main process. + ## Building and Running Node 24 is required to build and run this, as well as pnpm for dependency management. @@ -78,23 +133,24 @@ pnpm dev-workers There's a Dockerfile that can be used to build a container image. -The Kubernenetes resource files are in our private repository. +The Kubernetes resource files are kept in a private infrastructure repository. The Dockerfile doesn't contain a `CMD` instruction, so you need to specify the command to run when starting the container. -To build the Docker image, tagged both with a local name and one for the OCI registry, run: +To build the Docker image, tagged both with a local name and one for the container registry, run: ```bash +export REGISTRY= && \ export DATE_TAG=$(date +%s).$(git rev-parse --short HEAD) && \ echo $DATE_TAG && \ docker build --platform linux/amd64 -f ./Dockerfile \ --tag archived-repositories-checker:"${DATE_TAG}" \ - --tag sjc.ocir.io/axbydjxa5zuh/archived-repositories-checker:"${DATE_TAG}" . + --tag "${REGISTRY}"/archived-repositories-checker:"${DATE_TAG}" . ``` -To push the Docker image to the OCI registry, run: +To push the Docker image to the container registry, run: ```bash -docker push sjc.ocir.io/axbydjxa5zuh/archived-repositories-checker:"${DATE_TAG}" +docker push "${REGISTRY}"/archived-repositories-checker:"${DATE_TAG}" ``` diff --git a/services/libs/tinybird/README.md b/services/libs/tinybird/README.md index 4689fc40e9..bd4585e79c 100644 --- a/services/libs/tinybird/README.md +++ b/services/libs/tinybird/README.md @@ -4,6 +4,9 @@ - [Introduction](#introduction) - [Journey of Data](#journey-of-data-from-cdp-to-insights) +- [Architecture Overview](#architecture-overview) +- [Code Contributions and Collaborations](#code-contributions-and-collaborations) +- [CDP Activity Queries](#cdp-activity-queries-activities_relations_filtered) - [Making Changes to Resources](#making-changes-to-resources) - [How to Iterate on Data](#how-to-iterate-on-data) - [Testing Tinybird Pipes Locally](#testing-tinybird-pipes-locally) @@ -73,6 +76,52 @@ The following table compares the two parallel architectures processing activityR - **Use Lambda Architecture output** (`activityRelations_enriched_deduplicated_bucket_union`) for: CDP operations, monitoring, any use case requiring complete unfiltered data, where we can not use the filtered buckets +## Code contributions and collaborations + +Insights metrics count activities by class: code contributions, collaborations, or other. The +classes come from the `activityTypes` table, and each metric pipe either takes the classes as +parameters through `activityTypes_filtered.pipe` or fixes them in its SQL. See +[contributions-and-collaborations.md](./contributions-and-collaborations.md) for the table, the +parameters and the list of pipes in each group. + +## CDP activity queries: `activities_relations_filtered` + +The CDP backend lists and counts activities through `pipes/activities_relations_filtered.pipe`. +The caller is `queryActivities` and `queryActivitiesCounter` in +`services/libs/data-access-layer/src/activities/sql.ts`; `buildActivitiesParams` in +`services/libs/data-access-layer/src/activities/tinybirdAdapter.ts` builds the parameters. + +**Source.** The pipe reads Lambda output, so results include bots and every repository. When +`G1_memberIds` is set and no `G2_` to `G5_` parameter is, it reads `activityRelations_by_member_ds`, a member-sorted +mirror of `activityRelations_enriched_deduplicated_bucket_union` that +`activityRelations_by_member_copy.pipe` rebuilds. Any other filter shape reads the bucket +union directly. The mirror is a full snapshot (`COPY_MODE replace`, daily at 03:35 UTC), so it normally lags the bucket union by up to a day. New activities are missing from member-only queries until the next copy; existing rows keep their old relation fields, and removed rows can still appear. Content fields (`attributes`, `url`, `body` and `title`) come from `activities_deduplicated_ds` at query time and can reflect newer changes. + +**Base filters.** `segments`, `startDate`, `endDate`, `platform`, `activity_type`, +`activity_types`, `repos` and `indirectFork` apply to every row. `repos` and `indirectFork` look +up the activity in `activities_deduplicated_ds`: `repos` by a case-insensitive substring match on +`url`, `indirectFork` by the `isIndirectFork` attribute. The pipe applies no +contribution filter of its own. + +**Filter groups.** For anything the base filters cannot express, the pipe takes up to five groups, +`G1_` to `G5_`. Each group accepts `memberIds`, `activityTypes`, `organizationIds`, `platforms`, +`channels` and `ids`, each with an `_exclude` variant. Conditions inside a group combine with AND; +groups combine with OR; the group block as a whole is ANDed with the base filters. + +```text +base filters AND ( (G1 conditions) OR (G2 conditions) OR ... OR (G5 conditions) ) +``` + +`buildActivitiesParams` turns the API's logical filter (`and`, `or`, `not`) into this shape. Each +branch of an `or` becomes a group, top-level `and` conditions are copied into every group, and +`not` maps to the `_exclude` parameters. The adapter drops groups past the fifth and logs a +warning. It also caps `segments` at 600 IDs. + +**Output.** Rows sort by `timestamp DESC, id DESC` and page with `page` and `pageSize`. A +`searchTerm` matches the activity `title` or `body` in `activities_deduplicated_ds`. +`countOnly=1` returns a single `count` row instead of the page. The final node joins `attributes`, +`url`, `body` and `title` from `activities_deduplicated_ds` for the rows on the page. + ## Making changes to resources 1. Install the **tb client** for classic tinybird @@ -144,7 +193,7 @@ ALTER TABLE public."tableName" REPLICA IDENTITY FULL; ``` 3. (only for PROD) You need to create the topic in oracle kafka, it doesn't get created automaticly -4. Update tinybird kafka connect plugin env ( it's under crowd-kube/lf-prod-oracle(lf-staging-oracle)/kafka-connect/tinybird-sink.properties.enc ), there are list of tracked files in the decrypted file. +4. Add the topic to the Tinybird sink configuration of Kafka Connect, in the `crowd-kube` repository. That configuration lists the tracked topics. 5. Restart kafka-connect 6. Create tinybird datasource schema and push it to tinybird 7. Add access to the table to sequin user in postgres diff --git a/services/libs/tinybird/contributions-and-collaborations.md b/services/libs/tinybird/contributions-and-collaborations.md new file mode 100644 index 0000000000..f5dae07a9c --- /dev/null +++ b/services/libs/tinybird/contributions-and-collaborations.md @@ -0,0 +1,119 @@ +# Code contributions and collaborations + +Insights metrics split activities into three classes: code contributions, collaborations, and +everything else. This doc covers where that classification lives and how the Tinybird pipes apply +it. The user-facing explanation of each metric lives with the Insights frontend docs. + +## The `activityTypes` table + +One row per `(activityType, platform)` pair in the Postgres table `activityTypes`: + +| Column | Meaning | +| -------------------- | ------------------------------------------------------------- | +| `activityType` | Matches `activityRelations.type`, for example `issues-opened` | +| `platform` | Matches `activityRelations.platform`, for example `github` | +| `isCodeContribution` | The activity counts as a code contribution | +| `isCollaboration` | The activity counts as a collaboration | +| `description` | What the activity represents | +| `label` | Text the Insights UI shows for the type | + +The two flags are independent. A row with both flags false is an "other" activity. + +Examples from the initial seed: `authored-commit` on `git` and `pull_request-opened` on `github` are +code contributions; `issues-opened` and `issue-comment` on `github` are collaborations. + +Rows come from Flyway migrations in `backend/src/database/migrations/`. The table and its first +rows are in `V1757490935__create_activityTypes_table.sql`; later migrations add rows when an +integration introduces new activity types. To classify a new type, add a migration that inserts +its row. + +The match key is the `(type, platform)` pair. An activity whose pair has no row in +`activityTypes` drops out of every metric that filters through the table, including the "other" +class. + +## How the table reaches Tinybird + +The same migration adds `activityTypes` to the `sequin_pub` publication, so the table follows the +standard replication path (Postgres, Sequin, Kafka, Kafka Connect) into the Tinybird datasource +`datasources/activityTypes.datasource`. That datasource is a `ReplacingMergeTree` sorted by +`platform, activityType`, versioned by `updatedAt`. See [dataflow.md](./dataflow.md) and +[sequin.md](./sequin.md) for the path. + +Metric pipes join on `activityTypes`. They do not read the `isContribution` column that +`activityRelations` carries. + +## How pipes apply the classification + +The metric pipes below read the cleaned buckets of the +[bucketing architecture](./bucketing-architecture.md), so bots and disabled repositories are +already gone before the classification applies. Repositories marked `excluded` are not removed +there: bucket cleaning uses `repos_to_channels`, which checks only `enabled`. Pipes that must +leave out excluded repositories filter them separately, for example with +`repos_to_channels_excluded` as `health_score_active_contributors` does; the aggregates without +that filter still count them. + +```mermaid +flowchart LR + pg[("Postgres: activityTypes")] --> tb[("Tinybird: activityTypes")] + tb --> atf["activityTypes_filtered"] + atf --> af["activities_filtered"] + af --> widgets["Widget pipes"] + atf --> hs["Health score pipes"] + tb --> copies["Copy pipes with fixed classes"] +``` + +### Caller-chosen classes: `activityTypes_filtered.pipe` + +`pipes/activityTypes_filtered.pipe` returns the `(activityType, platform)` pairs that match three +boolean parameters: + +| Parameter | Default | Selects rows where | +| --------------------------- | ------- | -------------------------------------------------- | +| `includeCodeContributions` | 1 | `isCodeContribution = 1` | +| `includeCollaborations` | 0 | `isCollaboration = 1` | +| `includeOtherContributions` | 0 | `isCodeContribution = 0` and `isCollaboration = 0` | + +The conditions combine with OR. With no parameters the pipe returns code contributions only; the +caller widens the set by passing the other flags. + +Pipes that filter with `(type, platform) IN (SELECT activityType, platform FROM activityTypes_filtered)` +inherit these parameters from the request: + +- `activities_filtered.pipe`, `activities_filtered_historical_cutoff.pipe` and + `activities_filtered_retention.pipe`. The widget pipes built on `activities_filtered` are + `active_contributors`, `active_organizations`, `active_days`, `activities_count`, + `contributors_leaderboard`, `organizations_leaderboard`, `contributors_geo_distribution` and + `organizations_geo_distribution`. +- The health score pipes that count contributors: `health_score_active_contributors`, `health_score_contributor_dependency`, + `health_score_organization_dependency`, `health_score_retention`, and their `repo_health_score_*` + counterparts. +- `cncf_country_activities.pipe` and `cncf_country_comparison.pipe`. + +`activityTypes_by_project.pipe` takes the same three parameters and returns only the types that +have activities in the project or collection, with their `label`, for UI filter lists. + +### Fixed classes in copy pipes + +These pipes hard-code the classification in their SQL instead of taking parameters. + +Code contributions or collaborations (`isCodeContribution = 1 OR isCollaboration = 1`): + +| Pipe | Computes | +| --------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------- | +| `segmentId_aggregates_snapshot`, `segmentId_aggregates_initial_snapshot` | Distinct `contributorCount` and `organizationCount` per segment, into `segmentsAggregatedMV` | +| `collection_insights_aggregate_copy_pipe_0` to `_9` | Unique contributors per collection, one pipe per collection bucket | +| `leaderboards_members`, `leaderboards_organizations` | Activity count per member or organization, last 12 months against the 12 before | +| `leaderboards_project_active_contributors`, `leaderboards_project_active_organizations` | Unique contributors or organizations per project, same two periods | +| `agentic_ai_momentum_glance_copy`, `agentic_ai_projects_list_copy` | Agentic AI collection metrics | + +Code contributions only (`isCodeContribution`): + +| Pipe | Computes | +| -------------------------------------------- | -------------------------------------------------------------- | +| `org_page_kpis_copy_pipe` | Organization KPIs, including active contributors over 730 days | +| `org_page_projects_copy_pipe` | Activity and contributor counts per organization and project | +| `org_page_activities_timeseries_copy_pipe` | Yearly code contribution counts per organization | +| `org_page_contributors_timeseries_copy_pipe` | Yearly unique contributors per organization | + +To change what a fixed-class metric counts, edit the `WHERE` clause in its pipe. To change which +activities fall in a class, change the `activityTypes` rows with a migration.