[PROPOSAL] API Endpoint for Supervisor Errors

[justinborromeo]

Motivation (see #6571)

Currently, there's a status API endpoint that allows users to retrieve the status of a supervisor and its tasks. However, if errors are occurring, users have no way of determining what exceptions are being thrown without digging into log messages. If there was an API endpoint that returned the error-level exceptions for a specific supervisor, diagnosing issues would be much easier.

Analysis of Options

The two design decisions that need to be made are how the exceptions will be stored and the design of the API.

Storage

1. In-Memory: Write each logged exception to an on-heap circular buffer on the leading overlord with a configurable maximum of elements.
2. Database (use metadata store): Write every logged exception to some sort of system table. Periodically run a task to purge old events.
3. Druid? It would be neat to be able to perform Druid queries such (especially SQL) on log data.

API

1. Return all the stored exceptions (the number stored is configurable)
2. Return the last m types of exceptions with the timestamps of their last n occurrences

Proposed Changes and Rationale

For the sake of simplicity, I propose using the in-memory storage approach. The one disadvantage with this approach is that it doesn't let users perform post-mortems if the Overlord goes down (unlike the database approach). This likely won't be a significant disadvantage because the logs can always be analyzed in the event of Overlord failure.

Added Configs:
druid.kafka.ingestion.numLoggedErrorsStoredPerSupervisor: The number of error log messages to store in memory per supervisor. Config value is an int.

Also for the sake of simplicity, the following API endpoints will return the stored error log messages:

Get errors from a specific supervisor: GET /druid/indexer/v1/supervisor/<supervisorId>/errors

{
  "supervisorId":_______________,
 "errors":[
    {
      "timestamp":_______________,
      "errorMessage":_______________,
      "supervisorState":________________,
    },
    {
      "timestamp":_______________,
      "errorMessage":_______________,
      "supervisorState":________________,
    }
  ]
}

Bulk-get errors from all supervisors: GET /druid/indexer/v1/supervisor/errors

{
  "supervisorErrors": [
    {
      "supervisorId":_______________,
      "errors":[
        {
          "timestamp":_______________,
          "errorMessage":_______________,
          "supervisorState":________________,
        },
        {
          "timestamp":_______________,
          "errorMessage":_______________,
          "supervisorState":________________,
        }
      ]
    }
  ]
}

I plan to achieve this behaviour by extending EmittingLogger and adding an error(Throwable t, String message, String supervisorId) method to call Logger#error() then write details about the exception to a CircularBuffer in the corresponding supervisor class. Then, an additional errors method would be added to SupervisorResource to create an endpoint that returns the contents of the corresponding buffer.

Operational impact

No significant operational impact.

[jihoonson]

Hmm, I'm not sure my comment is directly related to this proposal.. Maybe it is somehow related. Anyway, I'd like to leave some note.

I understand this API can be used for UIs to show users something happened, but not sure how much it would be useful for now. In many cases, I feel the supervisor errors are not much self-descriptive. Exceptions show something happened, but it wasn't easy for me to figure out why it happened only from those exceptions. Rather, I should grep all logs of the supervisor from the overlord log file and look them through to guess what happened.

So, as well as this issue, I think we also need 1) an easy way to retrieve supervisor logs, or 2) make supervisor logs more descriptive. I'm not sure there's a good way for 1) with log4j though.

[gianm]

In-memory is a good way to go IMO. It's simple at least. And if we want to persist the error history at some point, we could layer that in the future.

I plan to achieve this behaviour by extending EmittingLogger and adding an error(Throwable t, String message, String supervisorId) method to call Logger#error() then write details about the exception to a CircularBuffer in the corresponding supervisor class.

How would this work -- how would the Logger find the right Supervisor object? I'm also not sure why it's necessary to modify the logger. Why not create a non-Logger-based method that does the same thing, that is part of the Supervisor itself and can be called directly from within the Supervisor code?

If there was a API endpoint that returned the error-level exceptions for a specific supervisor, diagnosing issues would be much easier.

I think this misses one important piece of information needed for diagnostics: what the supervisor was trying to do when it encountered the error, and whether the error is blocking progress or not. Probably the most common errors for the supervisor to run into are trouble connecting to the upstream system (Kafka/Kinesis). But it's important to know: was this the initial connection - meaning things are probably totally broken? Or was this not the initial connection, but a later one - meaning something that used to be working, now no longer works? Is it being retried or is it waiting for the user to do something? Also, is the error condition still ongoing? Or has it cleared? (Just looking at the most recent error won't tell you that - it also matters if the thing that encountered the error has succeeded in a later retry.)

[justinborromeo]

@jihoonson I definitely agree that this change isn't very useful for determining why something happened. I think the purpose outlined in the original issue was to let users know that something happened instead of silently failing and having users wonder why their supervisor isn't doing something.

However, I don't mind making changes to log descriptiveness; other than the info that @gianm mentioned, do you have any other ideas on what sort of information could be included to make it more descriptive? Streamlining the process of retrieving and searching through logs might be a good future task, though.

[justinborromeo]

@gianm

How would this work -- how would the Logger find the right Supervisor object? I'm also not sure why it's necessary to modify the logger. Why not create a non-Logger-based method that does the same thing, that is part of the Supervisor itself and can be called directly from within the Supervisor code?

My thinking was that the modified Logger could accept an injected TaskMaster object (which contains a SupervisorManager object) and retrieve the correct Supervisor object based on the provided supervisorId. However, I do agree that putting the storage logic in the Supervisor class would be a cleaner design.

I think this misses one important piece of information needed for diagnostics: what the supervisor was trying to do when it encountered the error, and whether the error is blocking progress or not. Probably the most common errors for the supervisor to run into are trouble connecting to the upstream system (Kafka/Kinesis). But it's important to know: was this the initial connection - meaning things are probably totally broken? Or was this not the initial connection, but a later one - meaning something that used to be working, now no longer works? Is it being retried or is it waiting for the user to do something? Also, is the error condition still ongoing? Or has it cleared? (Just looking at the most recent error won't tell you that - it also matters if the thing that encountered the error has succeeded in a later retry.)

I agree that those would be useful but how would you determine whether an exception is blocking progress? Also, by "what the supervisor was trying to do", do you mean the method the exception was thrown from or something higher-level?

[gianm]

I agree that those would be useful but how would you determine whether an exception is blocking progress? Also, by "what the supervisor was trying to do", do you mean the method the exception was thrown from or something higher-level?

I was imagining answering both of these the same way: by having a concept of what 'state' a supervisor is in, and inferring from that whether it's blocked or not and what it was trying to do.

[justinborromeo]

Proposed States:

POSTED_NOT_RUNNING_YET: Supervisor spec has been posted but waiting for start delay
TRYING_TO_CONTACT_KAFKA / KINESIS: Initially trying to connect to Kafka/Kinesis…hasn’t connected yet.
LOST_CONTACT_WITH_KAFKA / KINESIS: Trying to connect to Kafka/Kinesis…has connected before
TASKS_NOT_CREATED_YET: Supervisor started but hasn’t started indexing tasks yet
RUNNING: Started tasks and waiting for taskDuration to elapse
AWAITING_SHUTDOWN: Shutdown has been called but the supervisor hasn’t shutdown yet
NO_TASKS_RESPONDING: No tasks are responding to status requests

Does anyone have any ideas for other states?

[jon-wei]

I think NO_TASKS_RESPONDING is more of a kind of failure than a state (the state would be more why the supervisor was trying to contact tasks).

[justinborromeo]

Updated proposed states and when they change:

State	Description	Where supervisor gets set to this state
POSTED_NOT_RUNNING_YET	Supervisor spec has been posted but waiting for start delay	Upon SeekableStreamSupervisor instantiation
TRYING_TO_CONTACT_STREAM	Trying to connect to Kafka/Kinesis...hasn’t connected before	Upon the first throw of a BrokerNotFoundException (Kafka) or ResourceNotFoundException (Kinesis) if no record has been polled successfully yet.
LOST_CONTACT_WITH_STREAM	Trying to connect to Kafka/Kinesis...has connected before.	If a BrokerNotFoundException is thrown when polling Kafka for records or a ResourceNotFoundException is thrown when polling Kinesis for records.
UNABLE_TO_POLL_RECORDS	Non-connectivity related error occurring when polling for records	If any KafkaException that isn't a BrokerNotFoundException (for Kafka) or ResourceNotFoundException (for Kinesis) is thrown
TASKS_NOT_CREATED_YET	Supervisor started but hasn’t started indexing tasks yet	At the end of SeekableStreamSupervisor#tryInit()
RUNNING	Started tasks and waiting for taskDuration to elapse	At the end of SeekableStreamSupervisor#createNewTasks()
AWAITING_SHUTDOWN	Shutdown has been called but the supervisor hasn’t shutdown yet	At the beginning of SeekableStreamSupervisor#stop()
SUSPENDED	Supervisor is in a suspended state	At the beginning of SeekableStreamSupervisor#gracefulShutdownInternal()

[justinborromeo]

I was thinking of adding the exception and state data to the /v1/supervisor/{id}/status endpoint instead of creating a new endpoint. Would that cause issues from a compatibility point of view?

[dclim]

@justinborromeo some comments on how I would think about the supervisor state:

Talking about the happy path first:

I would define states which lets the user know where in the sequence of stages the supervisor is, but mainly during the first run. That's because most things happen badly in the first run, and some operations might take a long time, such as contacting the stream, trying to contact the discovered tasks to get their status, etc., where things are taking a while and the user wants to know what the supervisor is currently trying to do. Once that first run is complete, things should be somewhat more smooth, so I think we could keep it in a 'working normally' state until it's not working normally anymore. We can define 'not working normally' as having a certain threshold of exceptions that are thrown on consecutive runs; it might be cool if this is configurable too, because some people may be working in 'flakier' environments than others.

I think of it as: if it transitions from RUNNING (healthy) to UNHEALTHY, assume that someone has hooked up a monitoring system to it and is going to get paged at 4am in the morning, so it better actually be UNHEALTHY, and not some transient error that is going to resolve in the next minute. I would define thresholds for number of exceptions that you get back before you decide you're now unhealthy and someone needs to deal with you immediately - the threshold could be lower (or 1) for exceptions you know are never going to recover, and could be something higher for things that are likely to recover. Anything related to contacting tasks - such as timeouts waiting for the HTTP response or a bad response that causes you to kill the task, or the task was doing the wrong thing and was killed, to me doesn't constitute UNHEALTHY for the supervisor - it was unhealthy for the task, but the supervisor is designed to recover from this so that's cool, and someone should not get woken up for that unless it happens to multiple tasks consecutively and no ingestion progress can be made.

Suggest the following happy path states:

WAITING_TO_RUN (I like this a bit better than POSTED_NOT_RUNNING_YET; initial state)

CONNECTING_TO_STREAM (right before updatePartitionDataFromStream(), only first iteration)

DISCOVERING_INITIAL_TASKS (right before discoverTasks(), only first iteration)

CREATING_TASKS (right before createNewTasks())

RUNNING (after first iteration completes, and remain in this state until some unhealthy threshold is hit)

SUSPENDED (once we've become suspended - actually when we return, maybe we should go back to the first state of CONNECTING_TO_STREAM -> DISCOVERING_INITAL_TASKS -> etc. instead of just to RUNNING in case we've been suspended for like a month and things have changed)

SHUTTING_DOWN (replaces AWAITING_SHUTDOWN, it's not really 'waiting', it's actively doing stuff - same place, when stop() is called)

There can be a long period of time while it's WAITING_TO_RUN before it actually does anything so it's worth a state; it could also take a long time trying to connect to a stream if the stream can't be reached and it's waiting for timeout; same for timeouts while trying to send messages to indexing tasks. All those should only be hit on the first run; once it hits the RUNNING state, it should stay in RUNNING. I'd probably suggest that if it goes to SUSPENDED state and then is restarted, it should again iterate through the states starting with CONNECTING_TO_STREAM since things might have changed a lot and there's a higher probability of failure after a restart from suspended.

For error states I'm thinking:

UNABLE_TO_CONNECT_TO_STREAM - I like a bit better than UNABLE_TO_CONTACT_STREAM, because you might be able to 'contact' the stream but can't 'connect' - due to version mismatch, authentication, topic not existing, etc.; used for never was able to connect

LOST_CONTACT_WITH_STREAM - have been able to connect in the past

UNHEALTHY - general catch-all for failures, could include some stream related ones

So if it's in CONNECTING_TO_STREAM stage, it'll stay in that stage until either a) it connects and moves on to DISCOVERING_INITIAL_TASKS, or b) some threshold of failures has happened, in which case it'll move to UNABLE_TO_CONNECT_TO_STREAM; so it might be in CONNECTING_TO_STREAM stage for a minute or two minutes or five minutes before moving to a failure state, depending on what the threshold before failure is.

Once it makes it to RUNNING stage, it'll remain there until: a) it's suspended -> SUSPENDED state; b) it's shutdown - SHUTTING_DOWN state; c) some threshold of failures happen, in which case it'll move to either 1) LOST_CONTACT_WITH_STREAM, if that's what happened, or 2) UNHEALTHY for everything else. That's when the status changes to unhealthy, maybe the /status endpoint returns an HTTP 500 or something, and people need to deal with the issue.

Another thing is that some systems require a certain number of 'healthy' events to happen before transitioning out of the UNHEALTHY state back to RUNNING (healthy), to avoid flip-flopping between healthy and unhealthy when the environment is just flaky enough to be sitting on the threshold. To me, it's probably a nice to have but not required.