magmalake

Getting started

Step by step

mojo-census by hand: the tools, a new project with Orca in it, the code the agent wrote, excerpted from the repository, and how to run it here and on AWS. To run the example as it is instead, clone it, install the two tools below, and skip to Run it locally: 

git clone https://github.com/magmalake/mojo-census && cd mojo-census

Install pixi and mojoshelf

pixi manages the project's environment, the Mojo compiler included. mojoshelf's CLI adds pixi shelf, which finds and adds tins: 

curl -fsSL https://pixi.sh/install.sh | sh
pixi global install --channel https://mojoshelf.org/channel mojoshelf

Create the project

A new project needs Modular's channel, pixi's build preview (tins build with pixi-build) and the compiler before it adds Orca. The linux-aarch64 platform is for the image that runs on AWS Graviton instances later: 

pixi init lake -c https://conda.modular.com/max -c conda-forge && cd lake
sed -i.bak 's/^\[workspace\]/[workspace]\npreview = ["pixi-build"]/' pixi.toml
pixi add "mojo-compiler==1.2.0"
pixi workspace platform add linux-aarch64   # for images on AWS Graviton
pixi shelf add orca-mojo                     # orca-mojo 0.3.2, with dista-mojo, iceberg-mojo, ...

pixi shelf add orca-mojo brings orca-mojo 0.3.2 and the tins it needs. It takes a few minutes and prints every tin's build log as it goes; it hasn't hung. mojo-census also adds postgresql and git from conda-forge (pixi add postgresql git), for the local database and for cloning repositories inside the image. Its layout is the one the magmalake-apps skill describes: programs in bin/, one file per pipeline in pipelines/, shared code in src/, built with -I src.

The code

The orca command

The project's orca command, which registers pipelines, runs the scheduler and serves the UI, is two lines of its own. Its file must not be named orca.mojo, which would hide the orca package:

bin/scheduler.mojo

from orca.command import orca_command

def main() raises:
    orca_command()

gharchive

The gharchive pipeline is one step, mapped over days from 2023-01-01, on a cron schedule. The compiler checks the step's input and output types against the pipeline. writes= declares the Iceberg table it commits to, and Hour is the Dista task that downloads one hour of the archive, gunzips it with deflate-mojo's gunzip and parses it:

pipelines/gharchive.mojo

comptime Hour = Task2[String, Int, Seq[RecordBatch], hour_task]
comptime GhTasks = Tasks[Hour]

comptime LoadDay = Step[
    "load_day", Day, Int64, load_day, Mapped, retries=2, writes=[EVENTS]
]
comptime GhArchive = Pipeline[
    "gharchive",
    LoadDay,
    tasks=GhTasks,
    partitions=Daily["2023-01-01"],
    schedule=Cron["0 3 * * *"],  # 03:00 UTC, for the day before
    workers=4,
]

The step keeps four hours in flight at a time and appends each one as it comes back, so a day of about 15 GB of JSON never sits in memory whole. Each append is a snapshot that Orca numbers, so a retried attempt finds the hours its earlier attempt committed with find_commit and skips them: 

def load_day(mut ctx: Ctx, d: Day) raises -> Int64:
    var day = String(d)
    var hours = parse_hours(getenv("GHARCHIVE_HOURS"))
    var table = ensure_table(ctx, EVENTS, events_schema(), events_spec())
    ref rt = ctx.runtime[GhTasks]()
    var window = max(1, Int(getenv("GHARCHIVE_PARALLEL", "4")))

    # Hours an earlier attempt of this step already committed come first in
    # its commit order; recognise them without fetching anything.
    var done = 0
    while done < len(hours) and find_commit(ctx, table) >= 0:
        _ = append(ctx, table, List[RecordBatch]())
        done += 1
    if done > 0:
        ctx.log(String(day, ": ", done, " hours already committed"))

    var refs = List[ObjectRef[Seq[RecordBatch]]]()
    var next = done
    while next < len(hours) and next - done < window:
        refs.append(rt.remote[Hour](day, hours[next]))
        next += 1
    var rows = Int64(0)
    for i in range(done, len(hours)):
        var got = rt.get(refs[i - done])
        rt.release(refs[i - done])
        if next < len(hours):
            refs.append(rt.remote[Hour](day, hours[next]))
            next += 1
        var n = 0
        for b in got.items:
            n += b.num_rows
        var id = append(ctx, table, got.items)
        rows += Int64(n)
        ctx.log(String(day, " hour ", hours[i], ": ", n, " events, snapshot ", id))
    ctx.log(String(day, ": ", rows, " events in ", len(hours) - done, " hours"))
    return rows

mojo_repos

mojo_repos is a daily sensor on gharchive.events: each day gharchive commits starts a run for that day. It collects candidate repositories from the day's events, checks them against GitHub, and appends the Mojo ones to mojo.repos, keyed by owner/name. Every answer goes into mojo.checked, so a repository is asked about once.

mojo_history

mojo_history is keyed: its partitions are the repo column of mojo.repos, 50 to a run, and as a sensor on that table, each snapshot mojo_repos commits starts runs for the repositories it adds. Its one step gathers a run's keys and sends every repository to a Dista task that clones it and reads its log:

pipelines/mojo_history.mojo

comptime Histories = Step[
    "histories",
    Seq[Name],
    Int,
    histories,
    Gather,
    retries=1,
    reads=[REPOS],
    writes=[COMMITS, MIGRATIONS],
]
comptime MojoHistory = Pipeline[
    "mojo_history",
    Histories,
    tasks=HistoryTasks,
    partitions=Keyed[REPOS, "repo"],  # owner/name
    keys_per_run=50,
    sensor=True,
    workers=4,
]

Because the step declares mojo.repos in reads=, Orca's lineage runs all the way back: orca lineage mojo.commits traces the histories to the mojo.repos snapshot that started them and the gharchive.events snapshot and run that one came from.

The console

The console is a flare app over mojo.repos, mojo.commits and mojo.migrations. Each page also answers as JSON under /api/:

bin/console.mojo

def main() raises:
    var r = Router()
    r.get("/", index)
    r.get("/versions", versions)
    r.get("/repos/:owner/:name", repo_page)
    r.get("/api/repos", index_json)
    r.get("/api/versions", versions_json)
    r.get("/api/repos/:owner/:name", repo_json)
    var addr = String("0.0.0.0:", getenv("PORT", "8080"))
    var server = HttpServer.bind(SocketAddr.parse(addr), ServerConfig())
    print("console on http://", addr, sep="")
    server.serve(r^)

A route is a function from a request to a response. This one lists, for each month, how many libraries moved to each Mojo version, with the versions in one order across Modular's renumbering, which tests/test_versions.mojo tests: 

def versions(req: Request) raises -> Response:
    try:
        var rows = versions_by_month()
        var h = String(
            "<h1>Mojo versions over time</h1><p class=mute>Each month, how many",
            " libraries moved to each Mojo version (from their lock file, or",
            " their manifest when there's no lock), in Mojo's own order across",
            " the renumbering: 24.x, 25.x, 0.25.x, 0.26.x, 1.x.</p>",
            "<table><tr><th>Month</th><th>Version</th><th class=num>Libraries</th></tr>",
        )
        for ref r in rows:
            h.write(
                "<tr><td>", r.month, "</td><td>", html_escape(r.version),
                "</td><td class=num>", r.repos, "</td></tr>",
            )
        h.write("</table>")
        return _page("Mojo versions", h^)
    except e:
        return internal_error(String(e))

Run it locally

Build everything inside the pixi environment and run the tests. The binaries also run from a plain shell, but Postgres and git come from the pixi environment, so the commands below use pixi run or pixi shell. 

pixi run build      # build/orca, build/{gharchive,mojo_repos,mojo_history}, build/console
pixi run test       # version ordering and history parsing

The pipelines run on their own without a database. GHARCHIVE_HOURS keeps the download to two hourly files of about 100 MB each, and MOJO_HISTORY_MAX_KB skips repositories bigger than 100 MB, such as modular/modular: 

export ORCA_WAREHOUSE=$PWD/build/warehouse
GHARCHIVE_HOURS=15,17 pixi run ./build/gharchive run --partitions 2024-06-12
pixi run ./build/mojo_repos run --partitions 2024-06-12
MOJO_HISTORY_MAX_KB=100000 pixi run ./build/mojo_history run --partitions @all
PORT=8080 pixi run ./build/console      # http://127.0.0.1:8080

To run them the way they run in production, with the scheduler, the sensors and the UI, start a Postgres under build/: 

pixi run initdb -D build/pg --auth=trust --username=postgres -E UTF8 --locale=C
pixi run pg_ctl -D build/pg -l build/pg.log -o "-k $PWD/build -c listen_addresses=" -w start
pixi run createdb -h "$PWD/build" -U postgres orca

scripts/env.sh points ORCA_DATABASE_URL at it, puts the warehouse and logs under build/ and sets the small slice of the archive. Register all three pipelines before loading any data, because a sensor only counts snapshots made after it was registered. Then backfill one day: 

pixi shell
. scripts/env.sh
for p in gharchive mojo_repos mojo_history; do ./build/orca locate build/$p --root .; done
./build/orca register ./build/gharchive ./build/mojo_repos ./build/mojo_history
./build/orca scheduler &                 # fires, senses, starts runs
./build/orca ui &                        # http://127.0.0.1:8270 (ORCA_UI_ADDR to move it)
./build/orca backfill gharchive 2024-06-12

The scheduler runs gharchive for the day, which starts mojo_repos for the same day, whose new repositories start mojo_history. You can watch it in the UI, or in build/logs/<pipeline>/<run>/, and the console from above shows the result.

Without a token GitHub allows 60 API requests an hour, and mojo_repos spends one to three per candidate. On a day with more candidates than that, the rest are recorded as unchecked and the next run checks them first. To check them for that day once the limit has reset, re-run the day: a plain backfill skips days that already succeeded, and --rerun queues them anyway. 

./build/orca backfill --rerun mojo_repos 2024-06-12

Set GITHUB_TOKEN to lift the limit. Re-running a day of gharchive appends its events again, since a day's partition isn't overwritten yet. Stop the scheduler and the UI with kill and Postgres with pixi run pg_ctl -D build/pg stop.

Deploy to AWS

orca aws deploys the project as one CloudFormation stack. Use orca-mojo 0.3.2 or later: with 0.3.0 and 0.3.1, orca aws push fails to build the image on Linux when a pipeline links flare's HTTP client beside dista, as gharchive and mojo_repos do. orca aws drives the aws and docker commands you already have, so sign in with aws login (or set AWS_PROFILE) and have Docker running. The stack's Availability Zone defaults to us-east-2a; in another Region pass one of its zones with --zone. If you signed up through AWS's new experience, your project's Region is in AWS Settings, under View all projects, Overview, Additional Info.

Inside pixi shell, create the stack, with the email that gets its budget alerts and a hostname each for Orca's UI and the console. Then point an A record for each hostname at the PublicAddress that status prints. On Cloudflare, set the records to "DNS only", so that Caddy on the instance can get certificates. The UI also answers at SslipUrl straight away. 

./build/orca aws up --email you@example.com \
    --hostname orca.example.com --console-hostname mojo.example.com
./build/orca aws status

Commit, then build the linux-aarch64 image and push it to the stack's registry. Put a GitHub token in Secrets Manager under orca/, register the pipelines from pipelines/ with on-demand workers that scale from none to four while tasks wait, give the token to mojo_repos, and start the console on its hostname: 

./build/orca aws push
aws secretsmanager create-secret --name orca/github-token --secret-string "$GITHUB_TOKEN"
./build/orca aws register --workers 0..4 gharchive mojo_history
./build/orca aws register --workers 0..4 --secret GITHUB_TOKEN=orca/github-token mojo_repos
./build/orca aws app console

push builds one image for everything: the five binaries (orca, console, gharchive, mojo_repos and mojo_history) and the shared libraries they link, on debian:bookworm-slim. For mojo-census it is 71.1 MB compressed in ECR and 294 MB unpacked.

Until orca's next release, two things can trip a first deploy:

  • Right after the first push, the control instance is still starting its containers. If register says orca-refresh: command not found or No such container: scheduler, wait a minute and run it again.
  • The image needs CA certificates where libcurl looks for them, which is under the build environment's prefix, not the system's. Until objectstore reads SSL_CERT_FILE, copy mojo-census's Dockerfile.orca into your project, and push builds it in place of orca's default. It is orca's default image plus one line, COPY --from=build /src/project/.pixi/envs/default/ssl /src/project/.pixi/envs/default/ssl, which keeps the build environment's certificate directory at the same path in the final image, where libcurl (objectstore) and OpenSSL (flare) were compiled to look.

On AWS, GHARCHIVE_HOURS and MOJO_HISTORY_MAX_KB aren't set, so runs take all 24 hours of a day and every repository. Runs get the token as a variable, and every log Orca writes masks its value. After a code change, commit, then run push, register and app console again.

The UI is public for reading, and its actions need a login. There's no orca aws command for users or backfills yet, so add a user and start the first month's backfill on the control instance, through Session Manager. The same orca backfill --rerun re-runs a day there: 

aws ssm start-session --target <ControlInstance>      # from orca aws status
sudo docker exec -i scheduler orca user add alice        # reads the password from stdin
sudo docker exec scheduler orca backfill gharchive 2025-01-01..2025-01-31

What it costs. The control instance, a t4g.large on demand with a 40 GB volume and an Elastic IP, runs all month whether or not anything runs; it holds Postgres, the scheduler, the UI, Caddy and the console. Each run starts on-demand instances of its own, a c7g.large head and c7g.xlarge workers unless you register other types, and --workers 0..4 keeps an idle run at no workers. Downloads from the public bucket that holds the lake are billed to you. The stack emails you at 80% of a monthly budget, 100 dollars unless you pass --budget. Check the current prices for your Region, and look at what the one-month backfill cost before you backfill from 2023. When you're done, empty the buckets and run ./build/orca aws down --yes.

The Orca docs cover pipelines, the scheduler and the UI in full, and the magmalake-apps skill's AWS reference lists what the stack holds and what to do when a step of the deploy fails.