JournalConsumer permissions and scope?

[vsoch]

The JournalConsumer seems to require a URI where the running user is the instance owner. E.g., when I try to instantiate using the system instance, I am denied:

# system instance on corona
handle = flux.Flux("local:///var/run/local")

# This works ok
consumer = flux.job.JournalConsumer(handle)
consumer.start()

# This is denied
consumer.poll()

PermissionError: [Errno 1] Request requires owner credentials

That is failing here:

flux-core/src/common/libflux/msg_handler.c

Line 314 in f9209c2

errmsg = "Request requires owner credentials";

And I'm wondering if it might be possible to have it work, but filter to events that are owned by me? I was trying to tweak the rpc to set some kind of filter or constraint for that (just throwing salami at the wall, no luck so far)

class JournalConsumer(flux.job.JournalConsumer):
    @property
    def request_payload(self):
        return {"full": False, "userid": os.getuid(), "constraints": {"userid": os.getuid()}}

consumer = JournalConsumer(handle)
consumer.start()
consumer.poll()

Is this impossible? if I am submitting jobs on the head node, is there a reason I can't receive my job events? The use case is Genesis and being able to run a local service that can do that. When I do flux start I can only see one node (the head node) and can't submit to the system instance. I can do flux start and submit to the system instance:

flux start
FLUX_URI=local:///var/run/flux/local flux submit hostname

And indeed my job is there, but then I think events for that job (given running the consumer under flux start) would not seen because they aren't under my current instance.

Any help would be appreciated - let me know what I might try! The goals are to:

1. Be able to submit jobs to the system instance
2. Be able to receive events for them to respond to

This all works under an allocation, but the request is for the above.

[grondo]

The job events journal is emitted by the job-manager and thus it only knows about jobs within the instance it is running.

The journal contains entries for all jobs, which is why permission is restricted to the instance owner. It may be possible to filter the events based on the requesting user, but I see a couple challenges there:

• The job manager is a potential performance bottleneck since it is in the middle of all critical state transitions and interactions between the scheduler, job execution system, accounting, etc. I'm not sure that adding a new service open to all users is a good idea there.
• There may be high impact if we get something wrong in the implementation and send data about other users to a non-instance-owner consumer of the journal

These two bullets are the main reasons why the job-manager events journal is instance owner for now. (not to say we couldn't attempt to add something like this, but I'm not sure the impact it would have on a multi-user instance with a lot of users making use of this interface)

A user can get events for their own jobs in a multiuser instance now, just not from a single interface unfortunately. You can see how this is done in the JobWatcher class.

The one piece that is missing I think is an interface to get notified when a new job is submitted by you. 🤔 You might have to poll the job-list service for new jobs (using the since parameter) until we implement something to do this.

Another temporary solution might be to write a new Python-based service that connects to Flux as the instance owner and consumes the journal. This script could offer its own service for other users to consume their own events, possibly even using the same JournalConsumer interface. This would offload work from the job manager and could be implemented now.

[vsoch]

I agree that we should not put more burden on the system instance (and hurt one of the features that makes flux exception) nor should we approach that security risk. I would guess that the desire to run on a login node is primarily for persistence of the service.

I'll take a look at JobWatcher in the interim and see if I can mock something up that would work for our needs. We are currently running the JournalConsumer in a thread that writes events to a database, and the agents (the AI/ML dudes) have will have access to query that when they need. I'm going to see if there is some easy way to send the job id (to add it to the watcher) while it is running. My understanding of watch is that you have to add a group of ids at once and then wait (watch) for them to finish. We'd have to either have separate threads for each (not ideal) or a persistent service that can receive and handle multiple jobs.

Another temporary solution might be to write a new Python-based service that connects to Flux as the instance owner and consumes the journal. This script could offer its own service for other users to consume their own events, possibly even using the same JournalConsumer interface. This would offload work from the job manager and could be implemented now.

This would be nice as a cluster service, but isn't there still the issue of exposing another user's jobs?

[vsoch]

Here is a simple "submit this job and watch" example I put together:

import flux
import flux.job
import time
import sys
import os

def test_watcher():
    print("Flux Watcher Test")    
    # This is the system instance on Corona
    h = flux.Flux("local:///run/flux/local")

    # Make a dummy jobspec
    jobspec = flux.job.JobspecV1.from_command(["/bin/sleep", "10"])

    # Log events, status, and add the job id prefix
    watcher = flux.job.JobWatcher(h, log_events=True, log_status=True, labelio=True)

    # Submit the job and add to watcher
    jobid = flux.job.submit(h, jobspec)    
    watcher.add_jobid(flux.job.JobID(jobid))

    # Run the handle reactor and exit
    h.reactor_run()
    sys.exit(watcher.exitcode)

if __name__ == "__main__":
    test_watcher()    

Notably, the watcher is blocking. So the ideal design to have something running and send new job ids to it would not work. The approach to do one thread per job gets out of hand/messy fairly quickly beyond a few demo jobs. But maybe we can do another timer that will periodically add job ids from a queue while the reactor is running? This seems to run for at least this dummy case:

import flux
import flux.job
import queue
import threading
import time

# Shared URI by thread and submission
FLUX_URI = "local:///run/flux/local"

def run_test():

    # Create a queue to add job ids to (received by thread)
    q = queue.Queue()
    handle = flux.Flux(FLUX_URI)

    # Dunno what args / kwargs are here, just allow anything
    def watcher_thread(*args, **kwargs):
        h = flux.Flux(FLUX_URI)
        watcher = flux.job.JobWatcher(h, log_status=True, log_events=True, labelio=True)

        def check_queue(handle, timer, revents, user_args):
            try:
                while not q.empty():
                    jid = q.get_nowait()
                    print(f"[Watcher Thread] Dynamically adding {jid}")
                    watcher.add_jobid(flux.job.JobID(jid))
            except Exception as e:
                print(f"Timer error: {e}")

        # after, callback, repeat, args
        t = h.timer_watcher_create(0.1, check_queue, repeat=0.1)
        t.start()

        print("[Watcher Thread] Reactor Running...")
        h.reactor_run()

    # Start watcher in background (we hope)
    t = threading.Thread(target=watcher_thread, daemon=True)
    t.start()

    # Wait a bit (pretend MCP server is doing stuff)
    time.sleep(2)

    # Submit a job and send to watcher thread
    jobspec = flux.job.JobspecV1.from_command(["/bin/sleep", "5"])
    jobid = flux.job.submit(handle, jobspec)
    print(f"Job submitted: {jobid}. Sending to queue...")
    q.put(str(jobid))

    # Keep alive to see output. This would be kept alive by the server
    time.sleep(10)
    print("Test Finished.")


if __name__ == "__main__":
    run_test()

So I can next try to move this into the MCP server. It would start the thread on the server start, and the submit request would need to return the job id and add it to the queue. Then instead of print, it would add to the database as it does now. So far this works on the system instance / head node of the cluster, so it hasn't failed yet!

[vsoch]

The one piece that is missing I think is an interface to get notified when a new job is submitted by you. 🤔 You might have to poll the job-list service for new jobs (using the since parameter) until we implement something to do this.

And yes, this is spot on. I will think about it. I would rather avoid polling (that's why the JournalConsumer was so nice) but it would work for a demo while we make something better.

[vsoch]

ok it seems to plug in:

The test on the right isn't working because I'm just using the FluxWatcher and writing to stdio. I think I'm going to subclass it and have a custom watcher that writes to the database, and test that.

[grondo]

I posted JobWatcher mainly as an example to see how it starts watching the job eventlog for each newly monitored job. If it is working for now, that's great! One thing to be aware of, however, is that the JobWatcher class also watches the exec eventlog and output for jobs, which may be more work than is necessary for this use case.

Maybe we need a way to initialize the JobWatcher class to not do the extra monitoring (which is designed for use in flux watch and flux submit --watch)

Or I can write up an example that just watches the main eventlog (if I find some time after Tuesday's release)

Notably, the watcher is blocking. So the ideal design to have something running and send new job ids to it would not work.

Hm, this should work, since JobWatcher.add_jobs() just starts watching the eventlog for newly added jobs. You should be able to add jobs while the watcher is running and it will start monitoring them (i.e. in a callback in the reactor). Did you try this and it didn't work?

I guess one problem is that the JobWatcher is going to stop itself once all the jobs it is currently monitoring are inactive. This is another reason to perhaps consider a different design, or a new mode of JobWatcher, for this use case.

[vsoch]

Hm, this should work, since JobWatcher.add_jobs() just starts watching the eventlog for newly added jobs. You should be able to add jobs while the watcher is running and it will start monitoring them (i.e. in a callback in the reactor). Did you try this and it didn't work?

I have this working ok!

I guess one problem is that the JobWatcher is going to stop itself once all the jobs it is currently monitoring are inactive. This is another reason to perhaps consider a different design, or a new mode of JobWatcher, for this use case.

This might be the problem I'm running into now - I'll work on it this morning and (hopefully) post an update.

[vsoch]

okay now I'm actually running into issues with having the callbacks run with asyncio threads. I'm not sure that is ever going to work. To step back, we really need the JournalConsumer functionality, but to work outside of an allocation. The problem is that we are going to have work that needs to be submit for some unknown number of resources, so we can't just make an allocation first. The JournalConsumer is only working in the context of an allocation due to permissions. The watcher design adds complexity because I can't run it easily alongside the mcp server with asyncio, it seems that I need to make an entire independent service for it, which adds a lot of complexity to deploying this. At best I have to make a service that is either polling, or receives job ids via some shared database. I don't like any of my options.

[vsoch]

I'm wondering if this proposed solution is too complex. E.g., the assumption is that we should store the status info of jobs in a separate database so its always accessible to the MCP functions / agents. But arguably we could just have them request the info with a job info function call at any point in time when it's needed.

[vsoch]

I'm thinking that we should actually have the JournalConsumer operating on the level of a job. E.g., the submit still comes in from somewhere and hits the mcp server running on a head node. We probably need to do that for now so the jobs submit can request any number of cluster resources. The job is submit, and the equivalent JournalConsumer service (that right now is running with the MCP server) is run as a service with the job, and then the journal consumer (which is in the scope of the job / batch / alloc / whatever is created) can see everything and write events to the same database that is accessible to the MCP agents. That also creates a separation of responsibility, meaning that each job entity has its own service and they can access a common database. In summary:

1. I am remote to a cluster. I submit a job to it.
2. The request goes through our TBA receiver and is submit on login node A by an agent via flux-mcp-server
3. The agent function knows to do submit "plus a wrapper to start the JournalConsumer for the instance"
4. The instance JournalConsumer writes to the user-scoped database (one service per job instance)

And then we don't stress the head node beyond running the server that handles the initial submit, and each job has its own service (and we aren't stressing one service with a ton of jobs) and we can use the JournalConsumer without making some special or new thing.

[vsoch]

Is there a way to start a job level user service? E.g., the equivalent of this batch:

#!/bin/bash

systemctl --user start flux-scribe
flux submit <job stuff> --waitable
flux job wait
systemctl --user stop flux-scribe

It's kind of like a job level prolog / epilog. What I can do now is a flux jobspec that runs a broker with a files directive with that batch script:

But I don't think the flux python SDK has an interface for flux job submit <jobspec.json> so I would use subprocess and then capture the job id. It would be neat to have analogous interfaces to flux.job but with flux.batch

import flux.batch

# One of these
jobspec = flux.batch.BatchJobspecV1.from_script("./batch.sh")
jobspec = flux.batch.BatchJobspecV1.from_command(["sleep", "10"])

# Add some wrapping thing (this could also just be in the script)
prolog = "systemctl --user start flux-scribe"
jobspec.add_prolog(prolog)

epilog = "systemctl --user stop flux-scribe"
jobspec.add_epilog(epilog)

# or maybe even better, if we have structure to understand user services (and start / stop as a wrapper, consistently)
jobspec.add_service("flux-scribe")

I kind of like that client interface - maybe I'll mock something to play around with. It would be janky in needing subprocess but could work.

[grondo]

You can also use modprobe to schedule these kinds of tasks before and after the execution of a batch or alloc instance.
This would allow the service to be launched from rc1 and shut down in rc3 of the instance startup.

For example, a package might install the following to rc1.d/flux-scribe.py:

from flux.modprobe import task

@task(
    "start-flux-scribe",
    ranks="0",
    needs_config=["flux-scribe"],
    after=["resource", "job-list"],
)
def start_flux_scribe(context):
    context.bash("systemctl --user start flux-scribe")
    # or subprocess.rexec_bg(context.handle, ["/path/to/flux-scribe"], label="flux-scribe", waitable=True, nodeid=0)

and likewise in rc3.d/flux-scribe.py:

from flux.modprobe import task

@task(
    "stop-flux-scribe",
    ranks="0",
    needs_config=["flux-scribe"],
    before=["resource", "job-list"],
)
def stop_flux_scribe(context):
    context.bash("systemctl --user stop flux-scribe")
    # or 
    # subprocess.kill(handle, signum=2, label="flux-scribe").get()
    # status = subprocess.wait(handle, label="flux-scribe").get()["status"]

Then just add flux-scribe=true to your conf when creating jobspec to get the flux-scribe service.

[vsoch]

Oh, neat! I didn't know I could do that.

I'm going to add this (with user services) to https://github.com/converged-computing/flux-batch.

[vsoch]

OK - I'm mostly done with the service support. There are two issues (that we I think are aware of, could work on soon?) that prevent this from working nicely. The first is the lack of an ability to easily get/add the flux Python bindings to the PYTHONPATH. On Corona I usually need to do a find command to export PYTHONPATH and now I'm hitting this issue with my service:

I know this issue has been logged #7340. The second issue is the inability to interact with systemctl --user services within a typical batch job. What we have to do is start something in the background:

flux alloc -N 1 --bg

And then issue the start command to it:

ssh <node> systemctl --user start flux-scribe

Ideally we can just have the service start command in the batch job. Are we very far from being able to support that? Note that I'm testing on Corona. I'll take a look at adding the module support next - fingers crossed that works or I've ventured down another path that doesn't work (yet).