Feature/add new snapshot trigger and fix collector startup

[c0t0ber]

Added: New trigger has been added to initiate a snapshot for the collector based on time or the number of rows
fixed: Collector litestar server cant start #1012
adde

Also I can't get mypy to run, I've tried Python versions 3.8 and 3.10

  File "mypy\main.py", line 95, in main
  File "mypy\main.py", line 174, in run_build
  File "mypy\build.py", line 187, in build
  File "mypy\build.py", line 270, in _build
  File "mypy\build.py", line 2867, in dispatch
  File "mypy\build.py", line 3251, in process_graph
  File "mypy\build.py", line 3372, in process_stale_scc
  File "mypy\build.py", line 2442, in write_cache
  File "mypy\build.py", line 1559, in write_cache
  File "mypy\nodes.py", line 386, in serialize
  File "mypy\nodes.py", line 3649, in serialize
  File "mypy\nodes.py", line 3581, in serialize
AssertionError: Definition of pydantic.types.JsonValue is unexpectedly incomplete

Automatic generation of the Linestar API specification also does not work.

[provinzkraut]

The canonical way of handling long-running tasks in Litestar is to make use of the lifespan context managers, which are purpose built for this:

https://docs.litestar.dev/latest/usage/applications.html#lifespan-context-managers

You could define one like so:

Suggested change

	async def on_startup(app: Litestar) -> None:
	# TODO: check task health
	await loop(seconds=service.check_interval, func=check_snapshots_factory(service, service.storage))
	@contextlib.asynccontextmanager
	def lifespan(app: Litestar) -> AsyncGenerator[None, None]:
	func = check_snapshots_factory(service, service.storage)
	event = asyncio.Event()
	async def _loop(app: Litestar):
	while not event.is_set():
	await func()
	await asyncio.sleep(service.check_interval)
	task = asyncio.create_task(_loop())
	try:
	yield
	finally:
	event.set()
	await task

and add it to Litestar(..., lifespan=[lifespan]).

Litestar guarantees that things you call there won't be cancelled and will gracefully shutdown, even in case of a server error. The on_startup hook, while it can be used to achieve similar things, doesn't explicitly guarantee this and doesn't offer a way to handle graceful shutdowns of the tasks because it's a one-off function, which means you'll end up cancelling your tasks, which can lead to some warnings or potential errors because of unfinished coroutines. It's not a big deal in this case but I thought I'd mention it :)

Alternatively, if you don't want to deal with managing the task manually at all:

@contextlib.asynccontextmanager
def lifespan(app: Litestar) -> AsyncGenerator[None, None]:
    func = check_snapshots_factory(service, service.storage)

    async def _loop():
        while True:
            await func()
            await asyncio.sleep(service.check_interval)

    async with anyio.create_task_group() as tg:
      try:
          tg.start_soon(loop)
          yield 
      finally:
          tg.cancel_scope.cancel()

[mike0sv]

As for mypy, please ensure that you are using correct version from requirements.dev.txt. I. personally run it in 3.10 env.
As for api spec, there is a problem with Snapshot type hints i'm investigating. For now you can copy the approach from here

evidently/src/evidently/ui/api/projects.py

Line 286 in e27e366

snapshot = Snapshot.parse_raw(await request.body())

, replace snapshot model with request and parse it manually

[provinzkraut]

As for api spec, there is a problem with Snapshot type hints i'm investigating.

What is the issue exactly? If something is holding you guys up here don't hesitate to open an issue here or shoot us a message on our discord. It's always valuable for us to know what issues users are facing :)

---

Also, just a suggestion, but if you need the raw body you can inject it via the body kwarg (which also then allows you to just make a sync endpoint 🙂)

@post("/{project_id:uuid}/snapshots")
async def add_snapshot(
    project_id: Annotated[uuid.UUID, Parameter(title="id of project")],
    body: bytes,
    project_manager: Annotated[ProjectManager, Dependency()],
    log_event: Annotated[Callable, Dependency()],
    user_id: Annotated[UserID, Dependency()],
) -> None:
    snapshot = Snapshot.parse_raw(body)

[mike0sv]

@provinzkraut for now I know this. Snapshot model has Metric model as nested field somewhere (they are both pydantic). And Metric model has private field _context: "Context". It is annotated via string because of circular dependencies. Normally pydantic works with this because it ignores private fields. But when litestar tries to build schema, it tries to resolve it as ForwardRef and fails with NameError for "Context" (or something like that). I'll investigate further and probably can come up with small reproducible example

[mike0sv]

created an issue litestar-org/litestar#3150

[provinzkraut]

created an issue litestar-org/litestar#3150

Thanks, I'll look into it!

[provinzkraut]

@provinzkraut for now I know this. Snapshot model has Metric model as nested field somewhere (they are both pydantic). And Metric model has private field _context: "Context". It is annotated via string because of circular dependencies. Normally pydantic works with this because it ignores private fields. But when litestar tries to build schema, it tries to resolve it as ForwardRef and fails with NameError for "Context" (or something like that). I'll investigate further and probably can come up with small reproducible example

This has been fixed here. A patch release will be published soon!

[mike0sv]

@c0t0ber schema generation should be fixed with next litestar release, kudos to @provinzkraut

[provinzkraut]

Since this is running in a Litestar application, there's a slight benefit of having this function use the same executor as Litestar does (the executor will be re-used by Litestar), so you could just make use of litestar.concurrency.sync_to_thread:

Suggested change

	current = await run_in_thread_poll_executor(storage.get_and_flush, collector.id) # FIXME: sync function
	current = await sync_to_thread(storage.get_and_flush, collector.id) # FIXME: sync function

Goes for the other instances as well.

[c0t0ber]

@mike0sv

As for mypy, please ensure that you are using correct version from requirements.dev.txt. I. personally run it in 3.10 env

I tried running it on Windows and Linux on versions 3.10, 3.8 and I am getting this error. I perform the dependency installation using pip install -e ".[dev]" and pip install -r requirements.dev.txt according to the contributor guide
I also asked a friend to follow the steps from the contribution guide, and he encountered the same error.

[provinzkraut]

@c0t0ber This seems to be a mypy bug. The version used here is a bit outdated. Upgrading to 1.8.0 solved this issue for me.

[mike0sv]

We'll probably need to add tests to collector as well at some point 😅

[c0t0ber]

I'll try to add smoke tests

[c0t0ber]

@mike0sv

I think there are enough changes in this merge request. I suggest creating an issue for the collector's client refinement, writing tests for the collector, and updating the mypy version
Awaiting review

[mike0sv]

For some reason test for win with py3.10 fails. I restarted it multiple times in case its a timeout problem, but error persists. Do you have access to win machine to check it out?

[c0t0ber]

I found an issue with passing nan, inf, -inf values; msgpack cannot handle these values and fails on JSON parsing.

One of the solutions I came up with is to parse the request manually in all places where these values could potentially appear.

@provinzkraut
Is it possible to add a custom JSON decoder to litestar?

[provinzkraut]

Is it possible to add a custom JSON decoder to litestar?

Sure! You can extend the JSON parsing/serialisation by passing a custom encoding function for a type to the type_decoders available on all the app layers. For example:

def is_nan(v: Any) -> bool:
 ...

def decode_nan(value: Any) -> nan:
 ...

app = Litestar(
  type_decoders=[
    (is_nan, decode_nan)
 ]
)

That being said, msgspec does support support these values. If you're encoding them to their string representation (e.g. "nan"), it should also be able to decode them.

[mike0sv]

Hey @c0t0ber! We decided to skip those tests for now because we got other complaints about errors in collector, so we need to your fixes merged before next release :) We can work on tests in another PR if you like, and this one we will merge once CI passes

[mike0sv]

As for nan/inf issue, can you give us a reproducible example?

[c0t0ber]

@mike0sv

As for nan/inf issue, can you give us a reproducible example?

example here #1031