Connect reminder service to minder server to dispatch reminders

[Vyom-Yadav]

Summary

Provide a brief overview of the changes and the issue being addressed.
Explain the rationale and any background necessary for understanding the changes.
List dependencies required by this change, if any.

Issue #2262 (Task - I)

Adds logic for sending reminders to the minder server. Minder server processes these reminders to trigger a TopicQueueReconcileRepoInit event to reconcile a repository.

Change Type

Mark the type of change your PR introduces:

• Bug fix (resolves an issue without affecting existing features)
• Feature (adds new functionality without breaking changes)
• Breaking change (may impact existing functionalities or require documentation updates)
• Documentation (updates or additions to documentation)
• Refactoring or test improvements (no bug fixes or new functionality)

Testing

Outline how the changes were tested, including steps to reproduce and any relevant configurations.
Attach screenshots if helpful.

Tested locally (no unit tests)

Review Checklist:

• Reviewed my own code for quality and clarity.
• Added comments to complex or tricky code sections.
• Updated any affected documentation.
• Included tests that validate the fix or feature.
• Checked that related changes are merged.

cc @evankanderson

[Vyom-Yadav]

UpdateReminderLastSentById queries are part of a transaction and are only committed if the messages are published successfully. Still, the transaction could fail, which would result in reminders being sent, but the reminder_last_sent field won't be updated (better than false positive i.e. reminder_last_sent was updated but reminders weren't sent)

[jhrozek]

I think this is fine. Did you consider to split sending the messages into a separate function or return earlier if len(messages) == 0? This might be a personal preference but I find it easier to read code with fewer indentation levels.

Something like:

if len(messages) == 0 {
   return errorSlice
}

// the rest of the code

[evankanderson]

It seems like the reminder_last_sent field is mostly diagnostic, because we're actually ticking forward on the (in-memory) repository cursor, rather than using reminder_last_sent to actually dictate when repos are reconsidered. I think this is correct, but that suggests to me that strict update correctness is less necessary.

(I'm not arguing against keeping this data, but it seems like the actual mechanism that prevents re-evaluation is when the reminder is received and a rule evaluation is complete, which could be minutes after the reminder is sent on a particularly bad day.)

[Vyom-Yadav]

How do we want to export the metrics? And some thoughts on adding these metrics?

[jhrozek]

Hmm, I wonder what metrics would be useful as a minder operator.

About the histogram, I think a histogram that shows me how out of date (how much before a cutoff) a repo was when it was selected would be useful.

I think the metrics would be interesting to fine-tune the algorithm, did you see some other value in the metrics?

Since reminder is a separate service, then I guess implementation-wise it should just have its own metrics server that could be scraped separately.

[evankanderson]

My $0.02 would be:

1. Create a MeterProvider ala https://github.com/stacklok/minder/blob/main/internal/controlplane/server.go#L188.
2. Wire the metrics into prometheus. Since this has no external interface, you can simply set up an internal http.Server to serve the metrics -- we could put other debug information there later.
3. If you call otel.SetMeterProvider in the server setup, you don't need to worry about plumbing directly from the code recording the metric to the server; OTel will handle this in the backend (with some globals, but that's fine in this case). That means that the metrics code can do the simple thing from the OTel docs:

   var timeShift metrics.Float64Histogram
   func init() {
     meter := otel.GetMeterProvider().Meter("reminder")  // Or could be ""
     var err error
     timeShift, err = meter.Float64Counter("send_delay", api.WithDescription("Delay in revisit after eligible"), api.WithUnit("s"))
     if err != nil {
       panic("Couldn't register meter: %v", err)
     }
   }
   
   // At call site:
     timeShift.Record(ctx, send_delay)  // No need for attributes

[Vyom-Yadav]

I'll add monitoring in a sep PR.

[Vyom-Yadav]

Speaking about metrics, send_delay would be defined as:

	cutoff := time.Now().Add(-1 * r.cfg.RecurrenceConfig.MinElapsed)
	for _, repo := range repos {
		if t, ok := idToLastUpdate[repo.ID]; ok && t.Before(cutoff) {
			eligibleRepos = append(eligibleRepos, repo)
			// t is the last_update time for that repo
                        // send_delay ≈ time.Now().Sub(t) - r.cfg.RecurrenceConfig.MinElapsed
		}
	}

Now send_delay time has nothing to do with reminder_last_sent. What metrics do we want to record with reminder_last_sent? reminder_last_sent_delay isn't useful as the repo might not have been eligible in between so this doesn't represent the algorithm's performance. Struggling to find the value of reminder_last_sent as a metric parameter.

[evankanderson]

What I was thinking of recording on each pass of collecting reminders would effectively be the following for each reminded row:

now() - reminder_last_sent

Which would hopefully give a nice even distribution between MinElapsed and MinElapsed + Interval in normal operation. Unfortunately, common values might be MinElapsed = 24h and Interval = 5m, which will lead to a histogram where everything lands in a very large bucket. So, for practicality, I'd probably report this as:

now() - reminder_last_sent - MinElapsed

Which should give a range 0 .. Interval. If the algorithm starts to get behind or reminders "bunch", we'd see the tail of the histogram get bigger, and go beyond Interval. Given the target of Interval = 5m (hypothetical, but seems reasonable), we'd probably want to measure in seconds with a starting bucket size of 10s, roughly-doubling buckets (or a custom 1..5..10 bucketer) and going out to at least 20 minutes, which would mean about 9 buckets (0, 10, 20, 40, 80, ... 1280). I'm less concerned about the exact bucket sizing as being able to see whether most items are ending up within that MinElapsed + Interval window.

I'd store this metric in UpdateReminderLastSentForRepositories, so that the rest of the loop didn't need to care, and it was next to the handling of reminder_last_sent.

[coveralls]

coverage: 53.167% (-0.2%) from 53.4%
when pulling d480686 on Vyom-Yadav:send-reminder-events
into cd7a0f2 on stacklok:main.

[coveralls]

coverage: 53.172% (-0.2%) from 53.4%
when pulling d480686 on Vyom-Yadav:send-reminder-events
into cd7a0f2 on stacklok:main.

[coveralls]

coverage: 53.177% (-0.2%) from 53.4%
when pulling d480686 on Vyom-Yadav:send-reminder-events
into cd7a0f2 on stacklok:main.

[jhrozek]

The code looks quite good to me. I left replies to your comments, neither of them is blocking. It wasn't clear to me if you wanted to tackle the metrics as part of this PR, but I normally prefer smaller PRs.

[evankanderson]

It looks like the current architecture re-uses the existing events configuration for Minder, since it's assuming that the repo.reminder.event events are in a Watermill instance which Minder is configured to see. I think that's fine, but I didn't recall for sure where we'd landed on that earlier. This setup is 👍 for me from a simplicity PoV.

[evankanderson]

My preference would be to order these Project > Provider > Repo, which is the ownership order.

[Vyom-Yadav]

Done.

[evankanderson]

IMO, this should use the id field from the repositories table, not the repo_id from GitHub. That will help insulate this from GitHub-specific fields that are different in (for example) GitLab or BitBucket in the future.

[evankanderson]

Additionally, id is the primary key. It looks like our current indexes are set up off repo_id; sorry for not noticing this previously...

[evankanderson]

Looking further, it looks like we use the GitHub-matching repo_id in NewRepoReconcilerMessage already, so that argues towards using it here as well.

[Vyom-Yadav]

Yupp, I used RepositoryID to represent GH Id, as Id collided with internal primary key.

[evankanderson]

What's the benefit of calling validate and checking that the numeric repo_id is non-negative? It seems like the caller of this method will still need to check whether or not repo_id actually exists in the database, so this seems like it adds some extra code without really accomplishing the necessary validation; the necessary validation requires a database connection, and would presumably return the actual row if needed.

[Vyom-Yadav]

Yeah, no such benefit, removed it.

[evankanderson]

My $0.02 would be:

1. Create a MeterProvider ala https://github.com/stacklok/minder/blob/main/internal/controlplane/server.go#L188.
2. Wire the metrics into prometheus. Since this has no external interface, you can simply set up an internal http.Server to serve the metrics -- we could put other debug information there later.
3. If you call otel.SetMeterProvider in the server setup, you don't need to worry about plumbing directly from the code recording the metric to the server; OTel will handle this in the backend (with some globals, but that's fine in this case). That means that the metrics code can do the simple thing from the OTel docs:

   var timeShift metrics.Float64Histogram
   func init() {
     meter := otel.GetMeterProvider().Meter("reminder")  // Or could be ""
     var err error
     timeShift, err = meter.Float64Counter("send_delay", api.WithDescription("Delay in revisit after eligible"), api.WithUnit("s"))
     if err != nil {
       panic("Couldn't register meter: %v", err)
     }
   }
   
   // At call site:
     timeShift.Record(ctx, send_delay)  // No need for attributes

[evankanderson]

Do we need all of the following database reads and writes to go into a single transaction (but not the getRepositoryBatch call above in the transaction, or the r.updateRepositoryCursor in the outer getRepositoryBatch call)?

[Vyom-Yadav]

Removed the txn logic.

[evankanderson]

Intentional that you update when the reminder is sent before you send the message?

... except that with transactions, this actually sort-of happens after the message was sent, too -- you may be taking out a lock on the repo row for the duration of the iteration. Transactions can make before/after behavior hard to reason about in some cases, so I'd be cautious about introducing them when they aren't needed / aren't producing specific guarantees.

[Vyom-Yadav]

Changed this.

[evankanderson]

In general:

Either handle an error (by logging, recovering, etc) OR propagate an error, but don't do both.

Logging the error and also propagating it often means that you'll end up with errors being double- or even triple-logged, as different parts of the stack log the error but then pass it up to be logged in the next part of the stack.

[Vyom-Yadav]

Done.

[evankanderson]

Rather than an if/else here, generally prefer to use early-return, particularly since this seems like a terminal case.

More generally, I'm not convinced that there's a large benefit to batching the Publish messages, rather than sending each in the previous part of the loop.

[Vyom-Yadav]

Done.

I'm not convinced that there's a large benefit to batching the Publish messages, rather than sending each in the previous part of the loop.

Bulk updates might be better than connecting to the db for every update, I went with bulk updates with slight changes in logic.

[evankanderson]

It seems like the reminder_last_sent field is mostly diagnostic, because we're actually ticking forward on the (in-memory) repository cursor, rather than using reminder_last_sent to actually dictate when repos are reconsidered. I think this is correct, but that suggests to me that strict update correctness is less necessary.

(I'm not arguing against keeping this data, but it seems like the actual mechanism that prevents re-evaluation is when the reminder is received and a rule evaluation is complete, which could be minutes after the reminder is sent on a particularly bad day.)

[evankanderson]

It feels like this is simply forwarding messages from one queue to another; why not simply sending the TopicQueueReconcileRepoInit message directly from reminder.go and skip this extra intermediary? (There might be a good reason, but we should encapsulate that with a comment if there is such a reason.)

[Vyom-Yadav]

I'm open to removing this processor and directly sending events. But I think initially we discussed adding a new topic for reminder to have some distinction (which, now when I think cannot be internally distinguished, we don't have any form of tracing for reconciliations started by reminder).

[evankanderson]

In #4075 , I'm starting to look at switch from Watermill+SQL to CloudEvents + NATS, which has a somewhat more formalized envelope structure. I think we could probably have some sort of envelope property that distinguishes between the two, but put both messages onto the same topic -- the reader would be able to record which source the init event came from, but would generally handle "new" and "remind" the same.

[coveralls]

coverage: 53.818% (-0.2%) from 53.979%
when pulling f4968bd on Vyom-Yadav:send-reminder-events
into e10dba1 on stacklok:main.

[Vyom-Yadav]

Bulk updates for all repos is fine, IMO. We don't want to have a ms difference. The only value of this metric is in minutes/hours to make sure reconciliation is happening at a regular interval and we can reconcile every repo in like 24 hours or so.

[Vyom-Yadav]

We now stop if any JSON marshalling fails. I expect this to be deterministic, so if one is failing, others will fail too.

[Vyom-Yadav]

Sorry for the delay on this. Somehow, all weekends got occupied with one thing or the other :)

[evankanderson]

Sorry for the delay on this. Somehow, all weekends got occupied with one thing or the other :)

No problem, I've been in-and-out the last few weeks anyway. I'll try to take a look at this later today or tomorrow.

[Vyom-Yadav]

@evankanderson not found (404), where are you 😆

[evankanderson]

Sorry for disappearing! I had a week of vacation and a week of conferences, and when I got done, I had no idea what happened before that time.

This is looking pretty good, I'm happy to add it in as-is for now, and figure out the next steps in a subsequent PR. I did add a couple comments, but I don't think they block merging this.

[evankanderson]

Sorry for leaving this long enough that the underlying ground has changed a little, but you may want to make this an "entity id", so that we can remind about other things than repositories (artifacts, PRs, etc).

It's fine to do this in a subsequent PR if you prefer, since we'll still getting Reminder booted up.

[evankanderson]

It feels like it would make more sense to have a (e RepoReminderEvent) Marshal method, so you might use it as:

msg, err := RepoReminderEvent{projectID, providerID, repoID}.Marshal()

Since we don't need to validate any of the arguments to the message (it's just a data object), there's no need for a constructor.

[evankanderson]

I think this is okay as a constructor, though it only saves about 1 line of code elsewhere, because the caller will need to handle the error anyway.

[evankanderson]

What I was thinking of recording on each pass of collecting reminders would effectively be the following for each reminded row:

now() - reminder_last_sent

Which would hopefully give a nice even distribution between MinElapsed and MinElapsed + Interval in normal operation. Unfortunately, common values might be MinElapsed = 24h and Interval = 5m, which will lead to a histogram where everything lands in a very large bucket. So, for practicality, I'd probably report this as:

now() - reminder_last_sent - MinElapsed

Which should give a range 0 .. Interval. If the algorithm starts to get behind or reminders "bunch", we'd see the tail of the histogram get bigger, and go beyond Interval. Given the target of Interval = 5m (hypothetical, but seems reasonable), we'd probably want to measure in seconds with a starting bucket size of 10s, roughly-doubling buckets (or a custom 1..5..10 bucketer) and going out to at least 20 minutes, which would mean about 9 buckets (0, 10, 20, 40, 80, ... 1280). I'm less concerned about the exact bucket sizing as being able to see whether most items are ending up within that MinElapsed + Interval window.

I'd store this metric in UpdateReminderLastSentForRepositories, so that the rest of the loop didn't need to care, and it was next to the handling of reminder_last_sent.

[evankanderson]

In #4075 , I'm starting to look at switch from Watermill+SQL to CloudEvents + NATS, which has a somewhat more formalized envelope structure. I think we could probably have some sort of envelope property that distinguishes between the two, but put both messages onto the same topic -- the reader would be able to record which source the init event came from, but would generally handle "new" and "remind" the same.