fix(search): upsert ES updates and retry failed bulk indexes (#2523)

api/management/commands/index_and_notify.py

@@ -971,18 +971,41 @@ def index_records(self, records, to_create=True):
         self.bulk([construct_es_data(record, is_create=to_create) for record in list(records)])
 
     def bulk(self, actions):
+        if not actions:
+            return
         try:
             created, errors = bulk(client=ES_CLIENT, actions=actions)
             if errors:
                 logger.error(
                     "(index_and_notify:bulk): produced errors:",
-                    extra=logger_context(dict(errors=errors)),
+                    extra=logger_context(dict(errors=errors[:10], error_count=len(errors))),
                 )
+                # Retry failed updates individually with upsert (#2523)
+                for action in actions:
+                    if action.get("_op_type") != "update":
+                        continue
+                    try:
+                        bulk(client=ES_CLIENT, actions=[{**action, "doc_as_upsert": True}])
+                    except Exception:
+                        logger.error(
+                            "Failed to index record id=%s after bulk retry",
+                            action.get("_id"),
+                            exc_info=True,
+                        )
         except Exception:
             logger.error(
                 "(index_and_notify:bulk): could not index records",
                 exc_info=True,
             )
+            for action in actions:
+                try:
+                    bulk(client=ES_CLIENT, actions=[action])
+                except Exception:
+                    logger.error(
+                        "Failed to index record id=%s in fallback bulk",
+                        action.get("_id"),
+                        exc_info=True,
+                    )
 
     # Remove items in a queryset where updated_at == created_at.
     # This leaves us with only ones that have been modified.

utils/elasticsearch.py

@@ -42,24 +42,51 @@ def construct_es_data(instance, is_create=False):
         metadata.update(**data)
     else:
         metadata["doc"] = data
+        # Create the document when an update targets a missing ES record (#2523)
+        metadata["doc_as_upsert"] = True
     return metadata
 
 
+def _log_bulk_errors(errors, context=""):
+    if not errors:
+        return
+    logger.error(
+        "Elasticsearch bulk indexing produced errors%s: %s",
+        f" ({context})" if context else "",
+        errors[:5],
+        extra={"error_count": len(errors)},
+    )
+
+
 def create_es_index(instance):
     """Creates an Elasticsearch index from the record instance"""
 
     if ES_CLIENT and ES_PAGE_NAME:
-        # To make sure it doesn't run for tests
-        created, errors = bulk(client=ES_CLIENT, actions=[construct_es_data(instance, True)])
-        logger.info(f"Created {created} records")
-        log_errors(errors)
+        try:
+            created, errors = bulk(client=ES_CLIENT, actions=[construct_es_data(instance, True)])
+            logger.info(f"Created {created} records for {instance.__class__.__name__} pk={instance.pk}")
+            _log_bulk_errors(errors, f"create {instance.__class__.__name__}:{instance.pk}")
+        except Exception:
+            logger.error(
+                "Failed to index %s pk=%s",
+                instance.__class__.__name__,
+                instance.pk,
+                exc_info=True,
+            )
 
 
 def update_es_index(instance):
     """Updates the Elasticsearch index from the record instance"""
 
     if ES_CLIENT and ES_PAGE_NAME:
-        # To make sure it doesn't run for tests
-        updated, errors = bulk(client=ES_CLIENT, actions=[construct_es_data(instance)])
-        logger.info(f"Updated {updated} records")
-        log_errors(errors)
+        try:
+            updated, errors = bulk(client=ES_CLIENT, actions=[construct_es_data(instance)])
+            logger.info(f"Updated {updated} records for {instance.__class__.__name__} pk={instance.pk}")
+            _log_bulk_errors(errors, f"update {instance.__class__.__name__}:{instance.pk}")
+        except Exception:
+            logger.error(
+                "Failed to index %s pk=%s",
+                instance.__class__.__name__,
+                instance.pk,
+                exc_info=True,
+            )