diff --git a/.agents/AGENTS.md b/.agents/AGENTS.md index 1b6ac558e..746223e55 100644 --- a/.agents/AGENTS.md +++ b/.agents/AGENTS.md @@ -276,7 +276,8 @@ Processing services are FastAPI applications that implement the AMI ML API contr **Health Checks:** - Cached status with 3 retries and exponential backoff (0s, 2s, 4s) - Celery Beat task runs periodic checks (`ami.ml.tasks.check_processing_services_online`) -- Status stored in `ProcessingService.last_checked_live` boolean field +- Status stored in `ProcessingService.last_seen_live` boolean field +- Async/pull-mode services update status via `mark_seen()` when they register pipelines - UI shows red/green indicator based on cached status Location: `processing_services/` directory contains example implementations diff --git a/.agents/DATABASE_SCHEMA.md b/.agents/DATABASE_SCHEMA.md index 2a83bdb26..fdbbca832 100644 --- a/.agents/DATABASE_SCHEMA.md +++ b/.agents/DATABASE_SCHEMA.md @@ -255,8 +255,9 @@ erDiagram bigint id PK string name string endpoint_url - boolean last_checked_live - float last_checked_latency + datetime last_seen + boolean last_seen_live + float last_seen_latency } ProjectPipelineConfig { diff --git a/ami/jobs/views.py b/ami/jobs/views.py index 6d0626f23..832e15f30 100644 --- a/ami/jobs/views.py +++ b/ami/jobs/views.py @@ -30,6 +30,30 @@ logger = logging.getLogger(__name__) +def _mark_pipeline_pull_services_seen(job: "Job") -> None: + """ + Record a heartbeat for async (pull-mode) processing services linked to the job's pipeline. + + Called on every task-fetch and result-submit request so that the worker's polling activity + keeps last_seen/last_seen_live current. The periodic check_processing_services_online task + will mark services offline if this heartbeat stops arriving within PROCESSING_SERVICE_LAST_SEEN_MAX. + + IMPORTANT: This marks ALL async services on the pipeline within this project as live, not just + the specific service that made the request. If multiple async services share the same pipeline + within a project, a single worker polling will keep all of them appearing online. + Once application-token auth is available (PR #1117), this should be scoped to the individual + calling service instead. + """ + import datetime + + if not job.pipeline_id: + return + job.pipeline.processing_services.async_services().filter(projects=job.project_id).update( + last_seen=datetime.datetime.now(), + last_seen_live=True, + ) + + class JobFilterSet(filters.FilterSet): """Custom filterset to enable pipeline name filtering.""" @@ -245,6 +269,9 @@ def tasks(self, request, pk=None): if not job.pipeline: raise ValidationError("This job does not have a pipeline configured") + # Record heartbeat for async processing services on this pipeline + _mark_pipeline_pull_services_seen(job) + # Get tasks from NATS JetStream from ami.ml.orchestration.nats_queue import TaskQueueManager @@ -272,6 +299,9 @@ def result(self, request, pk=None): job = self.get_object() + # Record heartbeat for async processing services on this pipeline + _mark_pipeline_pull_services_seen(job) + # Validate request data is a list if isinstance(request.data, list): results = request.data diff --git a/ami/ml/migrations/0027_rename_last_checked_to_last_seen.py b/ami/ml/migrations/0027_rename_last_checked_to_last_seen.py new file mode 100644 index 000000000..4f14eee7c --- /dev/null +++ b/ami/ml/migrations/0027_rename_last_checked_to_last_seen.py @@ -0,0 +1,26 @@ +from django.db import migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ("ml", "0026_make_processing_service_endpoint_url_nullable"), + ] + + operations = [ + migrations.RenameField( + model_name="processingservice", + old_name="last_checked", + new_name="last_seen", + ), + migrations.RenameField( + model_name="processingservice", + old_name="last_checked_live", + new_name="last_seen_live", + ), + migrations.RenameField( + model_name="processingservice", + old_name="last_checked_latency", + new_name="last_seen_latency", + ), + ] diff --git a/ami/ml/models/pipeline.py b/ami/ml/models/pipeline.py index 2e47ffb86..c6d92e308 100644 --- a/ami/ml/models/pipeline.py +++ b/ami/ml/models/pipeline.py @@ -1043,7 +1043,7 @@ def online(self, project: Project) -> PipelineQuerySet: """ return self.filter( processing_services__projects=project, - processing_services__last_checked_live=True, + processing_services__last_seen_live=True, ).distinct() @@ -1142,7 +1142,7 @@ def collect_images( def choose_processing_service_for_pipeline( self, job_id: int | None, pipeline_name: str, project_id: int ) -> ProcessingService: - # @TODO use the cached `last_checked_latency` and a max age to avoid checking every time + # @TODO use the cached `last_seen_latency` and a max age to avoid checking every time job = None task_logger = logger @@ -1161,32 +1161,31 @@ def choose_processing_service_for_pipeline( # check the status of all processing services and pick the one with the lowest latency lowest_latency = float("inf") - processing_services_online = False + processing_service_lowest_latency = None for processing_service in processing_services: - if processing_service.last_checked_live: - processing_services_online = True - if ( - processing_service.last_checked_latency - and processing_service.last_checked_latency < lowest_latency - ): - lowest_latency = processing_service.last_checked_latency - # pick the processing service that has lowest latency + if processing_service.last_seen_live: + if processing_service.last_seen_latency and processing_service.last_seen_latency < lowest_latency: + lowest_latency = processing_service.last_seen_latency + processing_service_lowest_latency = processing_service + elif processing_service_lowest_latency is None: + # Online but no latency data (e.g. async/pull-mode service) — use as fallback processing_service_lowest_latency = processing_service - # if all offline then throw error - if not processing_services_online: + if processing_service_lowest_latency is None: msg = f'No processing services are online for the pipeline "{pipeline_name}".' task_logger.error(msg) - raise Exception(msg) - else: + + if lowest_latency < float("inf"): task_logger.info( f"Using processing service with latency {round(lowest_latency, 4)}: " f"{processing_service_lowest_latency}" ) + else: + task_logger.info(f"Using processing service (no latency data): {processing_service_lowest_latency}") - return processing_service_lowest_latency + return processing_service_lowest_latency def process_images( self, diff --git a/ami/ml/models/processing_service.py b/ami/ml/models/processing_service.py index ec7516d39..bad3dd147 100644 --- a/ami/ml/models/processing_service.py +++ b/ami/ml/models/processing_service.py @@ -22,8 +22,34 @@ logger = logging.getLogger(__name__) +# Max age of last_seen before a pull-mode (no-endpoint) service is considered offline. +# Pull-mode workers poll every ~5s, so 60s gives 12x buffer for transient failures. +PROCESSING_SERVICE_LAST_SEEN_MAX = datetime.timedelta(seconds=60) -class ProcessingServiceManager(models.Manager.from_queryset(BaseQuerySet)): + +class ProcessingServiceQuerySet(BaseQuerySet): + def async_services(self) -> "ProcessingServiceQuerySet": + """ + Filter to pull-mode (async) processing services — those with no endpoint URL. + + These correspond to jobs with dispatch_mode=ASYNC_API. Instead of Antenna calling + out to them, they poll Antenna for tasks and push results back. Their liveness is + tracked via heartbeats from mark_seen() rather than active health checks. + """ + return self.filter(models.Q(endpoint_url__isnull=True) | models.Q(endpoint_url__exact="")) + + def sync_services(self) -> "ProcessingServiceQuerySet": + """ + Filter to push-mode (sync) processing services — those with a configured endpoint URL. + + These correspond to jobs with dispatch_mode=SYNC_API. Antenna actively calls their + /readyz and /process endpoints. Their liveness is tracked by the periodic + check_processing_services_online Celery task. + """ + return self.exclude(models.Q(endpoint_url__isnull=True) | models.Q(endpoint_url__exact="")) + + +class ProcessingServiceManager(models.Manager.from_queryset(ProcessingServiceQuerySet)): """Custom manager for ProcessingService to handle specific queries.""" def create(self, **kwargs) -> "ProcessingService": @@ -41,12 +67,21 @@ class ProcessingService(BaseModel): projects = models.ManyToManyField("main.Project", related_name="processing_services", blank=True) endpoint_url = models.CharField(max_length=1024, null=True, blank=True) pipelines = models.ManyToManyField("ml.Pipeline", related_name="processing_services", blank=True) - last_checked = models.DateTimeField(null=True) - last_checked_live = models.BooleanField(null=True) - last_checked_latency = models.FloatField(null=True) + last_seen = models.DateTimeField(null=True) + last_seen_live = models.BooleanField(null=True) + last_seen_latency = models.FloatField(null=True) objects = ProcessingServiceManager() + @property + def is_async(self) -> bool: + """ + True if this is a pull-mode (async) service with no endpoint URL, corresponding to + jobs with dispatch_mode=ASYNC_API. False for push-mode services with a configured + endpoint, corresponding to jobs with dispatch_mode=SYNC_API. + """ + return not self.endpoint_url + def __str__(self): endpoint_display = self.endpoint_url or "async" return f'#{self.pk} "{self.name}" ({endpoint_display})' @@ -139,10 +174,27 @@ def create_pipelines( algorithms_created=algorithms_created, ) + def mark_seen(self, live: bool = True) -> None: + """ + Record that we heard from this processing service. + Used by async/pull-mode services that don't have an endpoint to check. + """ + self.last_seen = datetime.datetime.now() + self.last_seen_live = live + self.save(update_fields=["last_seen", "last_seen_live"]) + def get_status(self, timeout=90) -> ProcessingServiceStatusResponse: """ Check the status of the processing service. - This is a simple health check that pings the /readyz endpoint of the service. + + This check has two behaviors depending on the version of the processing service: + + If the service is a v2/pull-mode/async service with no endpoint URL, this will derive the status + from the last_seen heartbeat timestamp. If the last_seen timestamp is recent (within 60s), + the service is considered live. No requests are made by this method. + + If the service is a v1/push-mode/interactive service with an endpoint URL, this method will ping the + /readyz endpoint to check if it's live. Uses urllib3 Retry with exponential backoff to handle cold starts and transient failures. The timeout is set to 90s per attempt to accommodate serverless cold starts, especially for @@ -150,18 +202,28 @@ def get_status(self, timeout=90) -> ProcessingServiceStatusResponse: connection errors are handled gracefully. Args: - timeout: Request timeout in seconds per attempt (default: 90s for serverless cold starts) + timeout: Request timeout in seconds per attempt (default: 90s for serverless cold starts). Only applies \ + to services with an endpoint URL. """ - # If no endpoint URL is configured, return a no-op response - if self.endpoint_url is None: + # If no endpoint URL is configured, the derive status from last registration heartbeat + if not self.endpoint_url: + is_live = bool( + self.last_seen + and self.last_seen_live + and (datetime.datetime.now() - self.last_seen) < PROCESSING_SERVICE_LAST_SEEN_MAX + ) + if not is_live and self.last_seen_live: + # Heartbeat has expired — mark stale + self.last_seen_live = False + self.save(update_fields=["last_seen_live"]) + pipeline_names = list(self.pipelines.values_list("name", flat=True)) if is_live else [] return ProcessingServiceStatusResponse( - timestamp=datetime.datetime.now(), - request_successful=False, - server_live=None, - pipelines_online=[], + timestamp=self.last_seen or datetime.datetime.now(), + request_successful=is_live, + server_live=is_live, + pipelines_online=pipeline_names, pipeline_configs=[], - endpoint_url=self.endpoint_url, - error="No endpoint URL configured - service operates in pull mode", + endpoint_url=None, latency=0.0, ) @@ -171,7 +233,7 @@ def get_status(self, timeout=90) -> ProcessingServiceStatusResponse: pipeline_configs = [] pipelines_online = [] timestamp = datetime.datetime.now() - self.last_checked = timestamp + self.last_seen = timestamp resp = None # Create session with retry logic for connection errors and timeouts @@ -184,23 +246,23 @@ def get_status(self, timeout=90) -> ProcessingServiceStatusResponse: try: resp = session.get(ready_check_url, timeout=timeout) resp.raise_for_status() - self.last_checked_live = True + self.last_seen_live = True except requests.exceptions.RequestException as e: error = f"Error connecting to {ready_check_url}: {e}" logger.error(error) - self.last_checked_live = False + self.last_seen_live = False finally: latency = time.time() - start_time - self.last_checked_latency = latency + self.last_seen_latency = latency self.save( update_fields=[ - "last_checked", - "last_checked_live", - "last_checked_latency", + "last_seen", + "last_seen_live", + "last_seen_latency", ] ) - if self.last_checked_live: + if self.last_seen_live: # The specific pipeline statuses are not required for the status response # but the intention is to show which ones are loaded into memory and ready to use. # @TODO: this may be overkill, but it is displayed in the UI now. @@ -214,7 +276,7 @@ def get_status(self, timeout=90) -> ProcessingServiceStatusResponse: response = ProcessingServiceStatusResponse( timestamp=timestamp, request_successful=resp.ok if resp else False, - server_live=self.last_checked_live, + server_live=self.last_seen_live, pipelines_online=pipelines_online, pipeline_configs=pipeline_configs, endpoint_url=self.endpoint_url, @@ -229,7 +291,7 @@ def get_pipeline_configs(self, timeout=6): Get the pipeline configurations from the processing service. This can be a long response as it includes the full category map for each algorithm. """ - if self.endpoint_url is None: + if not self.endpoint_url: return [] info_url = urljoin(self.endpoint_url, "info") diff --git a/ami/ml/serializers.py b/ami/ml/serializers.py index 6c5782c8f..1711e19e1 100644 --- a/ami/ml/serializers.py +++ b/ami/ml/serializers.py @@ -2,6 +2,7 @@ from rest_framework import serializers from ami.main.api.serializers import DefaultSerializer, MinimalNestedModelSerializer +from ami.main.models import Project from .models.algorithm import Algorithm, AlgorithmCategoryMap from .models.pipeline import Pipeline, PipelineStage @@ -66,6 +67,8 @@ class Meta: class ProcessingServiceNestedSerializer(DefaultSerializer): + is_async = serializers.BooleanField(read_only=True) + class Meta: model = ProcessingService fields = [ @@ -73,8 +76,9 @@ class Meta: "id", "details", "endpoint_url", - "last_checked", - "last_checked_live", + "is_async", + "last_seen", + "last_seen_live", "created_at", "updated_at", ] @@ -134,6 +138,12 @@ class Meta: class ProcessingServiceSerializer(DefaultSerializer): pipelines = PipelineNestedSerializer(many=True, read_only=True) projects = serializers.SerializerMethodField() + is_async = serializers.BooleanField(read_only=True) + project = serializers.PrimaryKeyRelatedField( + write_only=True, + queryset=Project.objects.all(), + required=False, + ) class Meta: model = ProcessingService @@ -144,11 +154,13 @@ class Meta: "description", "projects", "endpoint_url", + "is_async", "pipelines", "created_at", "updated_at", - "last_checked", - "last_checked_live", + "last_seen", + "last_seen_live", + "project", ] def get_projects(self, obj): diff --git a/ami/ml/tasks.py b/ami/ml/tasks.py index 68e9603bd..3bfa458de 100644 --- a/ami/ml/tasks.py +++ b/ami/ml/tasks.py @@ -95,25 +95,49 @@ def remove_duplicate_classifications(project_id: int | None = None, dry_run: boo return num_deleted -@celery_app.task(soft_time_limit=10, time_limit=20) +# Timeout per sync service in the periodic beat task. Shorter than the default (90s for +# cold-start waits) since a missed check just waits for the next beat cycle. +# Worst case: 4 attempts (initial + 3 retries) × 8s timeout + backoff (0 + 2 + 4) = 38s per service. +_BEAT_STATUS_TIMEOUT = 8 + +# Discard queued copies that built up while the worker was unavailable — the next +# beat firing will pick things up fresh. Beat schedule is every 5 minutes. +_BEAT_TASK_EXPIRES = 240 + + +@celery_app.task(soft_time_limit=120, time_limit=150, expires=_BEAT_TASK_EXPIRES) def check_processing_services_online(): """ - Check the status of all v1 synchronous processing services and update the last_seen field. - We will update last_seen for asynchronous services when we receive a request from them. - - @TODO make this async to check all services in parallel + Check the status of all processing services and update last_seen/last_seen_live fields. + + - Async services (no endpoint URL): heartbeat is updated by mark_seen() on registration + and by _mark_pipeline_pull_services_seen() on task polling. This task marks them offline + if last_seen has exceeded PROCESSING_SERVICE_LAST_SEEN_MAX. Runs first so it always + executes even if a slow sync check hits the time limit. + - Sync services (endpoint URL set): checked sequentially with a short per-request timeout. + Safe to skip a cycle — the next beat firing will catch up. """ - from ami.ml.models import ProcessingService + import datetime - logger.info("Checking which synchronous processing services are online.") + from ami.ml.models.processing_service import PROCESSING_SERVICE_LAST_SEEN_MAX, ProcessingService - services = ProcessingService.objects.exclude(endpoint_url__isnull=True).exclude(endpoint_url__exact="").all() + logger.info("Checking which processing services are online.") - for service in services: - logger.info(f"Checking service {service}") + # Async services first — fast DB-only operation, must not be blocked by sync checks + stale_cutoff = datetime.datetime.now() - PROCESSING_SERVICE_LAST_SEEN_MAX + stale = ProcessingService.objects.async_services().filter(last_seen_live=True, last_seen__lt=stale_cutoff) + count = stale.count() + if count: + logger.info( + f"Marking {count} async service(s) offline (no heartbeat within {PROCESSING_SERVICE_LAST_SEEN_MAX})." + ) + stale.update(last_seen_live=False) + + for service in ProcessingService.objects.sync_services(): + logger.info(f"Checking push-mode service {service}") try: - status_response = service.get_status() + status_response = service.get_status(timeout=_BEAT_STATUS_TIMEOUT) logger.debug(status_response) - except Exception as e: - logger.error(f"Error checking service {service}: {e}") + except Exception: + logger.exception("Error checking service %s", service) continue diff --git a/ami/ml/tests.py b/ami/ml/tests.py index 642ca7131..36ba5b5f7 100644 --- a/ami/ml/tests.py +++ b/ami/ml/tests.py @@ -136,9 +136,9 @@ def test_create_processing_service_without_endpoint_url(self): # Check that endpoint_url is null self.assertIsNone(data["instance"]["endpoint_url"]) - # Check that status indicates no endpoint configured + # Check that status indicates service is not yet live (no heartbeat received) self.assertFalse(data["status"]["request_successful"]) - self.assertIn("No endpoint URL configured", data["status"]["error"]) + self.assertFalse(data["status"]["server_live"]) self.assertIsNone(data["status"]["endpoint_url"]) def test_get_status_with_null_endpoint_url(self): @@ -149,10 +149,8 @@ def test_get_status_with_null_endpoint_url(self): status = service.get_status() self.assertFalse(status.request_successful) - self.assertIsNone(status.server_live) + self.assertFalse(status.server_live) # No heartbeat received yet = not live self.assertIsNone(status.endpoint_url) - self.assertIsNotNone(status.error) - self.assertIn("No endpoint URL configured", (status.error or "")) self.assertEqual(status.pipelines_online, []) def test_get_pipeline_configs_with_null_endpoint_url(self): @@ -164,6 +162,118 @@ def test_get_pipeline_configs_with_null_endpoint_url(self): self.assertEqual(configs, []) +class TestProcessingServiceLastSeen(TestCase): + """Test the last_seen, last_seen_live, and last_seen_latency fields.""" + + def setUp(self): + self.project = Project.objects.create(name="Last Seen Test Project") + + def test_mark_seen_sets_fields(self): + """Test that mark_seen() sets last_seen and last_seen_live.""" + service = ProcessingService.objects.create(name="Async Worker", endpoint_url=None) + service.projects.add(self.project) + + self.assertIsNone(service.last_seen) + self.assertIsNone(service.last_seen_live) + + service.mark_seen(live=True) + service.refresh_from_db() + + self.assertIsNotNone(service.last_seen) + self.assertTrue(service.last_seen_live) + + def test_mark_seen_offline(self): + """Test that mark_seen(live=False) sets last_seen_live to False.""" + service = ProcessingService.objects.create(name="Async Worker Offline", endpoint_url=None) + + service.mark_seen(live=False) + service.refresh_from_db() + + self.assertIsNotNone(service.last_seen) + self.assertFalse(service.last_seen_live) + + def test_get_status_updates_last_seen_for_sync_service(self): + """Test that get_status() updates last_seen fields for sync services (even if endpoint is unreachable).""" + service = ProcessingService.objects.create(name="Sync Service", endpoint_url="http://nonexistent-host:9999") + service.projects.add(self.project) + + # get_status should update the fields even for unreachable endpoints + service.get_status(timeout=1) + service.refresh_from_db() + + self.assertIsNotNone(service.last_seen) + self.assertFalse(service.last_seen_live) # unreachable = not live + self.assertIsNotNone(service.last_seen_latency) + + def test_model_has_last_seen_fields(self): + """Test that ProcessingService model has last_seen fields and not last_checked.""" + service = ProcessingService.objects.create(name="Field Test Service", endpoint_url=None) + service.mark_seen(live=True) + service.refresh_from_db() + + # Verify new fields exist + self.assertTrue(hasattr(service, "last_seen")) + self.assertTrue(hasattr(service, "last_seen_live")) + self.assertTrue(hasattr(service, "last_seen_latency")) + + # Verify old fields don't exist + self.assertFalse(hasattr(service, "last_checked")) + self.assertFalse(hasattr(service, "last_checked_live")) + self.assertFalse(hasattr(service, "last_checked_latency")) + + +class TestProjectPipelineRegistrationUpdatesLastSeen(APITestCase): + """Test that async pipeline registration updates last_seen on the processing service.""" + + def setUp(self): + from ami.users.roles import ProjectManager, create_roles_for_project + + self.user = User.objects.create_user(email="lastseen@example.com") # type: ignore + self.project = Project.objects.create(name="Last Seen Project", owner=self.user, create_defaults=False) + create_roles_for_project(self.project) + ProjectManager.assign_user(self.user, self.project) + + def test_pipeline_registration_marks_service_as_seen(self): + """Test that POSTing to the pipeline registration endpoint marks the service as last_seen_live.""" + url = f"/api/v2/projects/{self.project.pk}/pipelines/" + payload = { + "processing_service_name": "AsyncTestWorker", + "pipelines": [], + } + + self.client.force_authenticate(user=self.user) + response = self.client.post(url, payload, format="json") + self.assertEqual(response.status_code, 201) + + service = ProcessingService.objects.get(name="AsyncTestWorker") + self.assertIsNotNone(service.last_seen) + self.assertTrue(service.last_seen_live) + + def test_repeated_registration_updates_last_seen(self): + """Test that re-registering updates the last_seen timestamp.""" + url = f"/api/v2/projects/{self.project.pk}/pipelines/" + payload = { + "processing_service_name": "AsyncTestWorkerRepeat", + "pipelines": [], + } + + self.client.force_authenticate(user=self.user) + + # First registration + self.client.post(url, payload, format="json") + service = ProcessingService.objects.get(name="AsyncTestWorkerRepeat") + first_seen = service.last_seen + + # Second registration + self.client.post(url, payload, format="json") + service.refresh_from_db() + second_seen = service.last_seen + + self.assertIsNotNone(first_seen) + self.assertIsNotNone(second_seen) + self.assertGreaterEqual(second_seen, first_seen) + + class TestPipelineWithProcessingService(TestCase): def test_run_pipeline_with_errors_from_processing_service(self): """ diff --git a/ami/ml/views.py b/ami/ml/views.py index b3272f567..58832a10b 100644 --- a/ami/ml/views.py +++ b/ami/ml/views.py @@ -277,4 +277,7 @@ def create(self, request, *args, **kwargs): projects=Project.objects.filter(pk=project.pk), ) + # Record that we heard from this async processing service + processing_service.mark_seen(live=True) + return Response(response.dict(), status=status.HTTP_201_CREATED) diff --git a/ui/src/data-services/models/pipeline.ts b/ui/src/data-services/models/pipeline.ts index 56b99d8b2..c5c676d31 100644 --- a/ui/src/data-services/models/pipeline.ts +++ b/ui/src/data-services/models/pipeline.ts @@ -78,7 +78,7 @@ export class Pipeline extends Entity { (service: any) => new ProcessingService(service) ) for (const processingService of processingServices) { - if (processingService.lastCheckedLive) { + if (processingService.lastSeenLive || processingService.isAsync) { return { online: true, service: processingService } } } @@ -90,7 +90,7 @@ export class Pipeline extends Entity { const processingServices = this._pipeline.processing_services let total_online = 0 for (const processingService of processingServices) { - if (processingService.last_checked_live) { + if (processingService.last_seen_live) { total_online += 1 } } @@ -98,22 +98,23 @@ export class Pipeline extends Entity { return total_online + '/' + processingServices.length } - get processingServicesOnlineLastChecked(): string | undefined { + get processingServicesOnlineLastSeen(): string | undefined { const processingServices = this._pipeline.processing_services if (!processingServices.length) { return undefined } - const last_checked_times = [] - for (const processingService of processingServices) { - last_checked_times.push( - new Date(processingService.last_checked).getTime() - ) + const last_seen_times = processingServices + .filter((s: any) => s.last_seen != null) + .map((s: any) => new Date(s.last_seen).getTime()) + + if (!last_seen_times.length) { + return undefined } return getFormatedDateTimeString({ - date: new Date(Math.max(...last_checked_times)), + date: new Date(Math.max(...last_seen_times)), }) } diff --git a/ui/src/data-services/models/processing-service.ts b/ui/src/data-services/models/processing-service.ts index 7f413a527..627a19616 100644 --- a/ui/src/data-services/models/processing-service.ts +++ b/ui/src/data-services/models/processing-service.ts @@ -1,4 +1,3 @@ -import { getFormatedDateTimeString } from 'utils/date/getFormatedDateTimeString/getFormatedDateTimeString' import { Entity } from './entity' import { Pipeline, ServerPipeline } from './pipeline' @@ -7,6 +6,7 @@ export type ServerProcessingService = any // TODO: Update this type export const SERVER_PROCESSING_SERVICE_STATUS_CODES = [ 'OFFLINE', 'ONLINE', + 'UNKNOWN', ] as const export type ServerProcessingServiceStatusCode = @@ -15,6 +15,7 @@ export type ServerProcessingServiceStatusCode = export enum ProcessingServiceStatusType { Success, Error, + Unknown, } export class ProcessingService extends Entity { @@ -36,22 +37,36 @@ export class ProcessingService extends Entity { return this._pipelines } - get endpointUrl(): string { - return `${this._processingService.endpoint_url}` + get id(): string { + return `${this._processingService.id}` } - get lastChecked(): string | undefined { - if (!this._processingService.last_checked) { + get name(): string { + return `${this._processingService.name}` + } + + get endpointUrl(): string | undefined { + const url = this._processingService.endpoint_url + return url && url.trim().length > 0 ? url : undefined + } + + get isAsync(): boolean { + return this._processingService.is_async ?? false + } + + get description(): string { + return `${this._processingService.description}` + } + + get lastSeen(): Date | undefined { + if (!this._processingService.last_seen) { return undefined } - - return getFormatedDateTimeString({ - date: new Date(this._processingService.last_checked), - }) + return new Date(this._processingService.last_seen) } - get lastCheckedLive(): boolean { - return this._processingService.last_checked_live + get lastSeenLive(): boolean { + return this._processingService.last_seen_live ?? false } get numPiplinesAdded(): number { @@ -64,7 +79,10 @@ export class ProcessingService extends Entity { type: ProcessingServiceStatusType color: string } { - const status_code = this.lastCheckedLive ? 'ONLINE' : 'OFFLINE' + if (this.isAsync) { + return ProcessingService.getStatusInfo('UNKNOWN') + } + const status_code = this.lastSeenLive ? 'ONLINE' : 'OFFLINE' return ProcessingService.getStatusInfo(status_code) } @@ -75,11 +93,13 @@ export class ProcessingService extends Entity { const type = { OFFLINE: ProcessingServiceStatusType.Error, ONLINE: ProcessingServiceStatusType.Success, + UNKNOWN: ProcessingServiceStatusType.Unknown, }[code] const color = { [ProcessingServiceStatusType.Error]: '#ef4444', // color-destructive-500, [ProcessingServiceStatusType.Success]: '#09af8a', // color-success-500 + [ProcessingServiceStatusType.Unknown]: '#9ca3af', // gray-400 }[type] return { diff --git a/ui/src/pages/processing-service-details/processing-service-details-dialog.tsx b/ui/src/pages/processing-service-details/processing-service-details-dialog.tsx index 1f8567bd4..895b820c9 100644 --- a/ui/src/pages/processing-service-details/processing-service-details-dialog.tsx +++ b/ui/src/pages/processing-service-details/processing-service-details-dialog.tsx @@ -78,8 +78,8 @@ const ProcessingServiceDetailsContent = ({ value={processingService.description} /> diff --git a/ui/src/pages/project/pipelines/pipelines-columns.tsx b/ui/src/pages/project/pipelines/pipelines-columns.tsx index 61b1231a4..384196ee9 100644 --- a/ui/src/pages/project/pipelines/pipelines-columns.tsx +++ b/ui/src/pages/project/pipelines/pipelines-columns.tsx @@ -63,11 +63,11 @@ export const columns: ( ), }, { - id: 'processing-services-online-last-checked', - name: 'Status last checked', - sortField: 'processing_services_online_last_checked', + id: 'processing-services-online-last-seen', + name: 'Status last seen', + sortField: 'processing_services_online_last_seen', renderCell: (item: Pipeline) => ( - + ), }, { diff --git a/ui/src/pages/project/processing-services/processing-services-columns.tsx b/ui/src/pages/project/processing-services/processing-services-columns.tsx index c33e6e40c..da2ab1078 100644 --- a/ui/src/pages/project/processing-services/processing-services-columns.tsx +++ b/ui/src/pages/project/processing-services/processing-services-columns.tsx @@ -1,3 +1,4 @@ +import { getFormatedDateTimeString } from 'utils/date/getFormatedDateTimeString/getFormatedDateTimeString' import { API_ROUTES } from 'data-services/constants' import { ProcessingService } from 'data-services/models/processing-service' import { BasicTableCell } from 'design-system/components/table/basic-table-cell/basic-table-cell' @@ -54,7 +55,13 @@ export const columns: ( renderCell: (item: ProcessingService) => ( ), diff --git a/ui/src/utils/language.ts b/ui/src/utils/language.ts index e1808b590..553f8b410 100644 --- a/ui/src/utils/language.ts +++ b/ui/src/utils/language.ts @@ -111,7 +111,7 @@ export enum STRING { FIELD_LABEL_JOB, FIELD_LABEL_JOBS, FIELD_LABEL_KEY, - FIELD_LABEL_LAST_CHECKED, + FIELD_LABEL_LAST_SEEN, FIELD_LABEL_LAST_DATE, FIELD_LABEL_LAST_SYNCED, FIELD_LABEL_LATITUDE, @@ -407,7 +407,7 @@ const ENGLISH_STRINGS: { [key in STRING]: string } = { [STRING.FIELD_LABEL_JOB]: 'Job', [STRING.FIELD_LABEL_JOBS]: 'Jobs', [STRING.FIELD_LABEL_KEY]: 'Key', - [STRING.FIELD_LABEL_LAST_CHECKED]: 'Last checked', + [STRING.FIELD_LABEL_LAST_SEEN]: 'Last seen', [STRING.FIELD_LABEL_LAST_DATE]: 'Last date', [STRING.FIELD_LABEL_LAST_SYNCED]: 'Last synced with data source', [STRING.FIELD_LABEL_LATITUDE]: 'Latitude',