class_docstr: A storage records all writing to storage sequentially.
code: def __init__(self, workflow_root_dir: str):
      super().__init__(workflow_root_dir)
      self._log_dir = self._workflow_root_dir
      self._count = self._log_dir / "count.log"
      self._op_counter = self._log_dir / "op_counter.pkl"
      if not self._log_dir.exists():
          self._log_dir.mkdir()
      # only one process initializes the count
      with FileLock(str(self._workflow_root_dir / ".lock")):
          if not self._count.exists():
              with open(self._count, "x") as f:
                  f.write("0")
          if not self._op_counter.exists():
              with open(self._op_counter, "wb") as f:
                  ray.cloudpickle.dump({}, f)
  
imports: import json
  from typing import Any, List
  from urllib import parse
  import pathlib
  from filelock import FileLock
  from ray.workflow.storage.base import Storage
  from ray.workflow.storage.filesystem import FilesystemStorageImpl
  import ray.cloudpickle
  from ray.workflow import serialization_context
label_desc: Use image node_id 1 to create a new LoggedStorage object from inherited base classes: FilesystemStorageImpl with example: obj = LoggedStorage(workflow_root_dir)