Status server for the filewriter and adding Logging #35

mortenjc · 2025-05-08T16:23:45Z

I added a status server to the filewriter

Since this is the first non EFU that reports a status a few changed were necessary:

type_fw = 5 was added
check_fw_pipeline() was added to parse the response from the server
check_service() was updated to handle fw
some formatting

milnowak · 2025-05-09T13:20:08Z

dashboard/generate.py

+    def check_fw_pipeline(self, ipaddr, port):
+        if self.test:
+            return 5
+        try:
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.connect((ipaddr, port))
+            s.send(b"getstatus")
+            data = s.recv(256)
+            data2 = int(data.decode("utf-8").split()[1][0])
+            s.close()
+            return data2
+        except:
+            self.dprint("connection reset (by peer?)")
+            return 0


I suggest switching to JSON for status messages from the services we poll in the dashboard and be more verbose with the status codes.

pros:

easier parsing on the Python side: we can access human readable keys directly as a dictionary, rather than splitting strings and working with indexes[0][1].

more specific error handling - for example, if the JSON is malformed, we get a JSONDecodeError. If a key is missing, we can catch a KeyError and handle it accordingly.

Right now, any issue in the try block is treated as a connection error, which can be misleading if the actual problem is with the message format.

class ServiceStatus(IntEnum): WORKING = 5 IDLE = 1 ERROR = 2 STOPPED = 3 INVALID = 4

def request_status(host, port): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((host, port)) s.send(b"getstatus") # Read until newline delimiter data = b"" while not data.endswith(b"\n"): chunk = s.recv(256) if not chunk: break data += chunk message = data.decode('utf-8').strip() try: return json.loads(message) except JSONDecodeError: logger.error(f"Error decoding JSON: {message}") return { "status": ServiceStatus.INVALID.value, "message": f"Didn't receive valid message from the service: {host}:{port}" }

then we can define check_status like that:

def check_status(self): response = request_status(host, port) try: status = ServiceStatus(int(response["status"])) except KeyError: logger.error(f"Status message must contain 'status' field. Ignoring message: {response}") except ValueError: logger.error(f"Status message contains invalid 'status' value. Valid status values: {list(ServiceStatus)}. Ignoring message: {response}") else: logging.debug("Setting new status: {self.idx} {status.name}") self.lab.setstatus(self.idx, status.value)

the debug message could look like this:

Setting new status: 127.0.0.1 WORKING

example message json:

{"status": 1, "message": "optional debug message"} # for filewriters {"status": 5, "stage1": 5, "stage2": 5, "stage2": 5} # for efus {"status": 2, "message": "Unhealthy", "uptime": "2d4h"} # anything else

If we agree on a common status message format for all services, we can reuse request_status and only implement a separate check_status function that extracts the necessary information from the JSON, depending on the service being polled.

I think this will make the code easier to read and debug.

Those ServiceStatus enumerations are what I was thinking of on the call @milnowak

I would suggest:
WRITING (green)
IDLE (green with lines)
STARTING (light green)
FINISHING (green)
FAILED (red)

and a status for OFFLINE (grey - response timed out; would need to be checked separately with ping like for EFUs)

Hopefully makes sense as a first pass from the meeting

milnowak · 2025-06-20T09:35:45Z

dashboard/generate.py

+            for line in lines:
+                try:
+                    obj = json.loads(line)
+                    # Look for any key ending with '.worker_state'
+                    for key, value in obj.items():
+                        if key.endswith(".worker_state"):
+                            return int(value)
+                except Exception:
+                    continue


Suggested change

for line in lines:

try:

obj = json.loads(line)

# Look for any key ending with '.worker_state'

for key, value in obj.items():

if key.endswith(".worker_state"):

return int(value)

except Exception:

continue

for line in lines:

obj = json.loads(line)

# Look for any key ending with '.worker_state'

for key, value in obj.items():

if key.endswith(".worker_state"):

return int(value)

this try/except block isn't necessary. When the line is {}, json.loads works fine and the inner for loop is simply skipped (no items, empty list). The only potential issue is a ValueError when converting to int but I’d rather let that fail. The error message will clearly show that the server sent something unexpected.

Suggested change

for line in lines:

try:

obj = json.loads(line)

# Look for any key ending with '.worker_state'

for key, value in obj.items():

if key.endswith(".worker_state"):

return int(value)

except Exception:

continue

for line in lines:

obj = json.loads(line)

# Look for any key ending with '.worker_state'

for key, value in obj.items():

if key.endswith(".worker_state"):

return int(value)

Thank you for the input! 🙌
I'll have it in mind going forward

I actually didn't want to have this to be reviewed since I'm yet to have it working (even with these changes). I'll have the PR as DRAFT until it's ready to be reviewed. Sorry not making it clear 😞

milnowak · 2025-06-20T09:54:13Z

dashboard/generate.py

+        except Exception:
+            self.dprint("connection reset (by peer?)")
+            return 0


I think in this case the exception handling might be hiding the real error and just assuming the connection was reset. It could make it harder to understand what actually went wrong.

We should at least handle OSError and ConnectionError

except ConnectionError: print("Connection reset by peer") except OSError as e: print(f"Socket error: {e}")

about workerState=0. does that mean the worker is idling? If the data is missing or couldn’t be parsed, maybe using something like -1 would make it clearer that the value is unknown, rather than a valid state.

ConnectionError is a subclas of OSError so catching only OSError is just fine.

Andre-Lx-Costa · 2025-06-20T12:31:20Z

WIP - Will be back from DRAFT as soon as it is ready 👍

ggoneiESS · 2025-06-24T13:42:21Z

@Andre-Lx-Costa if it helps, worker_state should always be last in the array

We can also modify the JSON so that we have a [{"key": "foo.bar", "value":"123"}, {...}] structure instead if it helps (or drop the square brackets entirely etc), if it makes it easier to handle. There is also an option to create something like the EFU handling where a specific request is sent and only certain keys are returned

This reverts commit 4e85d77.

ggoneiESS · 2025-06-30T21:18:32Z

LGTM. I think we should hold off merging until we understand the behaviour of the socket in the filewriter though, just in case we need to change something else in this to get good behaviour.

Andre-Lx-Costa · 2025-07-03T19:35:01Z

It is now ready to be reviewed 👍

We simplified the metrics being provided by the filewriter under the merge request here: https://gitlab.esss.lu.se/ecdc/ess-dmsc/kafka-to-nexus/-/merge_requests/915

The dashboard is deployed and the filewriter corresponding to the Merge Request above is the filewriter0625.daq.esss.dk

milnowak · 2025-07-04T07:50:53Z

dashboard/generate.py

+        try:
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.settimeout(3.0)
+            s.connect((ipaddr, port))
+            s.settimeout(20.0)
+            s.shutdown(socket.SHUT_WR)
+            data = b""
+            while True:
+                chunk = s.recv(4096)
+                if not chunk:
+                    break
+                data += chunk
+            text_data = data.decode("utf-8", errors="ignore").strip()
+            try:
+                parsed_json = json.loads(text_data)
+                for entry in parsed_json:
+                    for key, value in entry.items():
+                        if key.endswith(".worker_state"):
+                            return int(value)
+                return -1
+            except Exception as e:
+                self.dprint(f"JSON processing error: {e}")
+                return -1
+        except OSError as e:
+            self.dprint(f"Socket error: {e}")
+            return -1


you can catch multiple exceptions in one try block

I'd really suggest using a logger instead of print. print hides the stack trace and makes debugging tough. Use logger.exception and include context (ip:port).

return status objects instead of raw ints, it's way easier to read and avoids guesswork.

Suggested change

try:

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.settimeout(3.0)

s.connect((ipaddr, port))

s.settimeout(20.0)

s.shutdown(socket.SHUT_WR)

data = b""

while True:

chunk = s.recv(4096)

if not chunk:

break

data += chunk

text_data = data.decode("utf-8", errors="ignore").strip()

try:

parsed_json = json.loads(text_data)

for entry in parsed_json:

for key, value in entry.items():

if key.endswith(".worker_state"):

return int(value)

return -1

except Exception as e:

self.dprint(f"JSON processing error: {e}")

return -1

except OSError as e:

self.dprint(f"Socket error: {e}")

return -1

try:

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.settimeout(3.0)

s.connect((ipaddr, port))

s.settimeout(20.0)

s.shutdown(socket.SHUT_WR)

data = b""

while True:

chunk = s.recv(4096)

if not chunk:

break

data += chunk

parsed_json = json.loads(data.decode("utf-8", errors="ignore").strip())

for entry in parsed_json:

for key, value in entry.items():

if key.endswith(".worker_state"):

return Status(int(value))

return Status.SERVER_RUNNING

except JSONDecodeError as e:

logger.exception(f"Failed to load JSON response from Filewriter {ipaddr}:{port}")

return Status.SERVER_RUNNING

except OSError as e:

logger.exception(f"Failed to read from Filewriter socket {ipaddr}:{port}")

return Status.SERVER_RUNNING

Logger config

import logging logger = logging.getLogger(__name__) logger.setLevel(logging.DEBUG) formatter = logging.Formatter("[%(levelname)s] %(message)s") handler = logging.StreamHandler(sys.stdout) handler.setFormatter(formatter) logger.addHandler(handler)

from enum import Enum # adjust if needed class Status(Enum): INVALID = -1 SERVER_RUNNING = 0 SERVICE_RUNNING = 5

# switching between values and status objects >>> Status(-1) <Status.INVALID: -1> >>> Status(-1).value -1 >>> Status(5) <Status.SERVICE_RUNNING: 5> >>> Status.SERVICE_RUNNING.value 5

Good change but can we resolve that in a different PR? Because we need to change the whole script, not just for this section on the filewriter

I kept the dprints to keep the Repo's consistency.
If I am introducing the logger shouldn't I expand it for everything?

This dprint only prints if we're running the dashboard with debug as True so it won't even show up anywhere at the moment.

I do appreciate the suggestion and will implement it! But I think I have to do it for everything to keep some consistency across the other services

Thanks for the detailed suggestions 🙂
I've updated the PR now with 3c9a262

If I am introducing the logger shouldn't I expand it for everything?

it’s totally fine to take this step by step.
Refactoring the whole codebase isn't always feasible and honestly, we don’t know if anyone will even have time for that.

IMO, it's better to start improving things now rather than sticking to bad practices just to stay consistent.
Having this in place makes it easier to push for proper logging and status handling in future PRs. We can point to this as the starting point, and later extract the logger to a shared module for reuse across other components.

milnowak

great job! 🥇

dashboard/generate.py

Co-authored-by: Milosz Nowak <[email protected]>

mortenjc requested review from tbukovics, ggoneiESS and milnowak May 8, 2025 16:23

milnowak reviewed May 9, 2025

View reviewed changes

Andre-Lx-Costa force-pushed the fw_status_poc branch from c0ed4bc to bb9ff22 Compare June 20, 2025 08:26

milnowak requested changes Jun 20, 2025

View reviewed changes

Andre-Lx-Costa force-pushed the fw_status_poc branch from 1ac5926 to 50c53f2 Compare June 20, 2025 12:15

Andre-Lx-Costa marked this pull request as draft June 20, 2025 12:30

Andre-Lx-Costa self-assigned this Jun 20, 2025

Andre-Lx-Costa force-pushed the fw_status_poc branch 4 times, most recently from c18329d to 889b6c8 Compare June 24, 2025 12:52

mortenjc and others added 8 commits June 30, 2025 09:17

tested locally on my mac

7f5c688

updated csv with filewriter type and tcp port

81d00ba

update to use new filewriter status

5aaf474

Onboard filewriter new awesome metrics

9cab338

Revert "Onboard filewriter new awesome metrics"

276f799

This reverts commit 4e85d77.

Tweak fetching worker state metric

8aa2ad6

Milosz awesome comments

5d901bd

Attempt to better parse fw status

038c720

Andre-Lx-Costa force-pushed the fw_status_poc branch 7 times, most recently from 2a5d946 to 116fd49 Compare June 30, 2025 11:13

Andre-Lx-Costa force-pushed the fw_status_poc branch from 116fd49 to f378b03 Compare June 30, 2025 11:14

WIP - Tweak to find the last worker_state in the array

eace2f6

Andre-Lx-Costa force-pushed the fw_status_poc branch from f378b03 to eace2f6 Compare June 30, 2025 11:25

Cleanup

9591c19

Andre-Lx-Costa force-pushed the fw_status_poc branch from 1ce698f to 9591c19 Compare June 30, 2025 21:01

André Costa added 2 commits July 1, 2025 12:51

Convoluted solution but it works fine now

c3b5fc7

Back to basics - lets fix it on the filewriter

41ab917

Andre-Lx-Costa force-pushed the fw_status_poc branch from cab48fa to 41ab917 Compare July 3, 2025 13:48

Andre-Lx-Costa marked this pull request as ready for review July 3, 2025 19:35

milnowak requested changes Jul 4, 2025

View reviewed changes

Addressing PR comments

3c9a262

Andre-Lx-Costa requested a review from milnowak July 4, 2025 11:15

milnowak approved these changes Jul 4, 2025

View reviewed changes

dashboard/generate.py Outdated Show resolved Hide resolved

Update dashboard/generate.py

fe246ac

Co-authored-by: Milosz Nowak <[email protected]>

Andre-Lx-Costa changed the title ~~tested locally on my mac~~ Status server for the filewriter and adding Logging Jul 4, 2025

Andre-Lx-Costa merged commit c6e6d67 into master Jul 4, 2025

Status server for the filewriter and adding Logging #35

Status server for the filewriter and adding Logging #35

Uh oh!

Conversation

mortenjc commented May 8, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Andre-Lx-Costa commented Jun 20, 2025

Uh oh!

ggoneiESS commented Jun 24, 2025

Uh oh!

ggoneiESS commented Jun 30, 2025

Uh oh!

Andre-Lx-Costa commented Jul 3, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

milnowak Jul 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

milnowak left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

milnowak Jul 4, 2025 •

edited

Loading