Handle broken pipes when writing archive result. #1367

perfsonar · Feb 6, 2024 · 221316b · 221316b
1 parent 509481f
commit 221316b
Show file tree

Hide file tree

Showing 12 changed files with 59 additions and 14 deletions.
diff --git a/pscheduler-archiver-bitbucket/bitbucket/archive b/pscheduler-archiver-bitbucket/bitbucket/archive
@@ -10,6 +10,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER({"succeeded": True})
+    try:
+        EMITTER({"succeeded": True})
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
diff --git a/pscheduler-archiver-esmond/esmond/archive b/pscheduler-archiver-esmond/esmond/archive
@@ -296,6 +296,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
diff --git a/pscheduler-archiver-failer/failer/archive b/pscheduler-archiver-failer/failer/archive
@@ -71,6 +71,10 @@ for parsed in PARSER:
     if parsed["data"].get("badly", False):
         pscheduler.fail("Forced to fail badly.")
 
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
diff --git a/pscheduler-archiver-http/http/archive b/pscheduler-archiver-http/http/archive
@@ -205,6 +205,9 @@ EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 for parsed in PARSER:
     try:
         EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
     except Exception as ex:
         log.exception()
         EMITTER({

diff --git a/pscheduler-archiver-kafka/kafka/archive b/pscheduler-archiver-kafka/kafka/archive
@@ -188,6 +188,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
-pscheduler.succeed()
+pscheduler.succeed()
diff --git a/pscheduler-archiver-postgresql/postgresql/archive b/pscheduler-archiver-postgresql/postgresql/archive
@@ -193,6 +193,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
diff --git a/pscheduler-archiver-rabbitmq/rabbitmq/archive b/pscheduler-archiver-rabbitmq/rabbitmq/archive
@@ -287,6 +287,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
diff --git a/pscheduler-archiver-snmptrap/snmptrap/archive b/pscheduler-archiver-snmptrap/snmptrap/archive
@@ -183,6 +183,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
diff --git a/pscheduler-archiver-syslog/syslog/archive b/pscheduler-archiver-syslog/syslog/archive
@@ -127,6 +127,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
diff --git a/pscheduler-archiver-tcp/tcp/archive b/pscheduler-archiver-tcp/tcp/archive
@@ -119,7 +119,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
-
diff --git a/pscheduler-archiver-udp/udp/archive b/pscheduler-archiver-udp/udp/archive
@@ -118,7 +118,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()
-
diff --git a/scripts/PDK/templates/archiver/__PLUGIN_NAME__/archive b/scripts/PDK/templates/archiver/__PLUGIN_NAME__/archive
@@ -25,6 +25,10 @@ PARSER = pscheduler.RFC7464Parser(sys.stdin)
 EMITTER = pscheduler.RFC7464Emitter(sys.stdout)
 
 for parsed in PARSER:
-    EMITTER(archive(parsed))
+    try:
+        EMITTER(archive(parsed))
+    except BrokenPipeError as ex:
+        log.warning("Broken pipe during archiving; parent must have exited.")
+        pscheduler.succeed()
 
 pscheduler.succeed()