Loading tools/edit_monitor/Android.bp +21 −0 Original line number Diff line number Diff line Loading @@ -35,6 +35,12 @@ python_library_host { pkg_path: "edit_monitor", srcs: [ "daemon_manager.py", "edit_monitor.py", ], libs: [ "asuite_cc_client", "edit_event_proto", "watchdog", ], } Loading @@ -53,6 +59,21 @@ python_test_host { }, } python_test_host { name: "edit_monitor_test", main: "edit_monitor_test.py", pkg_path: "edit_monitor", srcs: [ "edit_monitor_test.py", ], libs: [ "edit_monitor_lib", ], test_options: { unit_test: true, }, } python_binary_host { name: "edit_monitor", pkg_path: "edit_monitor", Loading tools/edit_monitor/edit_monitor.py 0 → 100644 +125 −0 Original line number Diff line number Diff line # Copyright 2024, The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import getpass import logging import multiprocessing.connection import os import platform import time from atest.metrics import clearcut_client from atest.proto import clientanalytics_pb2 from proto import edit_event_pb2 from watchdog.events import FileSystemEvent from watchdog.events import PatternMatchingEventHandler from watchdog.observers import Observer # Enum of the Clearcut log source defined under # /google3/wireless/android/play/playlog/proto/log_source_enum.proto LOG_SOURCE = 2524 class ClearcutEventHandler(PatternMatchingEventHandler): def __init__( self, path: str, cclient: clearcut_client.Clearcut | None = None ): super().__init__(patterns=["*"], ignore_directories=True) self.root_monitoring_path = path self.cclient = cclient or clearcut_client.Clearcut(LOG_SOURCE) self.user_name = getpass.getuser() self.host_name = platform.node() self.source_root = os.environ.get("ANDROID_BUILD_TOP", "") def on_moved(self, event: FileSystemEvent): self._log_edit_event(event, edit_event_pb2.EditEvent.MOVE) def on_created(self, event: FileSystemEvent): self._log_edit_event(event, edit_event_pb2.EditEvent.CREATE) def on_deleted(self, event: FileSystemEvent): self._log_edit_event(event, edit_event_pb2.EditEvent.DELETE) def on_modified(self, event: FileSystemEvent): self._log_edit_event(event, edit_event_pb2.EditEvent.MODIFY) def flushall(self): logging.info("flushing all pending events.") self.cclient.flush_events() def _log_edit_event( self, event: FileSystemEvent, edit_type: edit_event_pb2.EditEvent.EditType ): event_time = time.time() logging.info("%s: %s", event.event_type, event.src_path) try: event_proto = edit_event_pb2.EditEvent( user_name=self.user_name, host_name=self.host_name, source_root=self.source_root, ) event_proto.single_edit_event.CopyFrom( edit_event_pb2.EditEvent.SingleEditEvent( file_path=event.src_path, edit_type=edit_type ) ) clearcut_log_event = clientanalytics_pb2.LogEvent( event_time_ms=int(event_time * 1000), source_extension=event_proto.SerializeToString(), ) self.cclient.log(clearcut_log_event) except Exception: logging.exception("Failed to log edit event.") def start( path: str, cclient: clearcut_client.Clearcut | None = None, pipe_sender: multiprocessing.connection.Connection | None = None, ): """Method to start the edit monitor. This is the entry point to start the edit monitor as a subprocess of the daemon manager. params: path: The root path to monitor cclient: The clearcut client to send the edit logs. conn: the sender of the pipe to communicate with the deamon manager. """ event_handler = ClearcutEventHandler(path, cclient) observer = Observer() logging.info("Starting observer on path %s.", path) observer.schedule(event_handler, path, recursive=True) observer.start() logging.info("Observer started.") if pipe_sender: pipe_sender.send("Observer started.") try: while True: time.sleep(1) finally: event_handler.flushall() observer.stop() observer.join() if pipe_sender: pipe_sender.close() tools/edit_monitor/edit_monitor_test.py 0 → 100644 +198 −0 Original line number Diff line number Diff line # Copyright 2024, The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Unittests for Edit Monitor.""" import logging import multiprocessing import os import pathlib import signal import sys import tempfile import time import unittest from atest.proto import clientanalytics_pb2 from edit_monitor import edit_monitor from proto import edit_event_pb2 class EditMonitorTest(unittest.TestCase): @classmethod def setUpClass(cls): super().setUpClass() # Configure to print logging to stdout. logging.basicConfig(filename=None, level=logging.DEBUG) console = logging.StreamHandler(sys.stdout) logging.getLogger('').addHandler(console) def setUp(self): super().setUp() self.working_dir = tempfile.TemporaryDirectory() self.root_monitoring_path = pathlib.Path(self.working_dir.name).joinpath( 'files' ) self.root_monitoring_path.mkdir() self.log_event_dir = pathlib.Path(self.working_dir.name).joinpath('logs') self.log_event_dir.mkdir() def tearDown(self): self.working_dir.cleanup() super().tearDown() def test_log_edit_event_success(self): fake_cclient = FakeClearcutClient( log_output_file=self.log_event_dir.joinpath('logs.output') ) p = self._start_test_edit_monitor_process(fake_cclient) # Create and modify a file. test_file = self.root_monitoring_path.joinpath('test.txt') with open(test_file, 'w') as f: f.write('something') # Move the file. test_file_moved = self.root_monitoring_path.joinpath('new_test.txt') test_file.rename(test_file_moved) # Delete the file. test_file_moved.unlink() # Give some time for the edit monitor to receive the edit event. time.sleep(1) # Stop the edit monitor and flush all events. os.kill(p.pid, signal.SIGINT) p.join() logged_events = self._get_logged_events() self.assertEqual(len(logged_events), 4) expected_create_event = edit_event_pb2.EditEvent.SingleEditEvent( file_path=str( self.root_monitoring_path.joinpath('test.txt').resolve() ), edit_type=edit_event_pb2.EditEvent.CREATE, ) expected_modify_event = edit_event_pb2.EditEvent.SingleEditEvent( file_path=str( self.root_monitoring_path.joinpath('test.txt').resolve() ), edit_type=edit_event_pb2.EditEvent.MODIFY, ) expected_move_event = edit_event_pb2.EditEvent.SingleEditEvent( file_path=str( self.root_monitoring_path.joinpath('test.txt').resolve() ), edit_type=edit_event_pb2.EditEvent.MOVE, ) expected_delete_event = edit_event_pb2.EditEvent.SingleEditEvent( file_path=str( self.root_monitoring_path.joinpath('new_test.txt').resolve() ), edit_type=edit_event_pb2.EditEvent.DELETE, ) self.assertEqual( expected_create_event, edit_event_pb2.EditEvent.FromString( logged_events[0].source_extension ).single_edit_event, ) self.assertEqual( expected_modify_event, edit_event_pb2.EditEvent.FromString( logged_events[1].source_extension ).single_edit_event, ) self.assertEqual( expected_move_event, edit_event_pb2.EditEvent.FromString( logged_events[2].source_extension ).single_edit_event, ) self.assertEqual( expected_delete_event, edit_event_pb2.EditEvent.FromString( logged_events[3].source_extension ).single_edit_event, ) def test_log_edit_event_fail(self): fake_cclient = FakeClearcutClient( log_output_file=self.log_event_dir.joinpath('logs.output'), raise_log_exception=True, ) p = self._start_test_edit_monitor_process(fake_cclient) # Create a file. self.root_monitoring_path.joinpath('test.txt').touch() # Give some time for the edit monitor to receive the edit event. time.sleep(1) # Stop the edit monitor and flush all events. os.kill(p.pid, signal.SIGINT) p.join() logged_events = self._get_logged_events() self.assertEqual(len(logged_events), 0) def _start_test_edit_monitor_process( self, cclient ) -> multiprocessing.Process: receiver, sender = multiprocessing.Pipe() # Start edit monitor in a subprocess. p = multiprocessing.Process( target=edit_monitor.start, args=(str(self.root_monitoring_path.resolve()), cclient, sender), ) p.daemon = True p.start() # Wait until observer started. received_data = receiver.recv() self.assertEquals(received_data, 'Observer started.') receiver.close() return p def _get_logged_events(self): with open(self.log_event_dir.joinpath('logs.output'), 'rb') as f: data = f.read() return [ clientanalytics_pb2.LogEvent.FromString(record) for record in data.split(b'\x00') if record ] class FakeClearcutClient: def __init__(self, log_output_file, raise_log_exception=False): self.pending_log_events = [] self.raise_log_exception = raise_log_exception self.log_output_file = log_output_file def log(self, log_event): if self.raise_log_exception: raise Exception('unknown exception') self.pending_log_events.append(log_event) def flush_events(self): delimiter = b'\x00' # Use a null byte as the delimiter with open(self.log_output_file, 'wb') as f: for log_event in self.pending_log_events: f.write(log_event.SerializeToString() + delimiter) self.pending_log_events.clear() if __name__ == '__main__': unittest.main() Loading
tools/edit_monitor/Android.bp +21 −0 Original line number Diff line number Diff line Loading @@ -35,6 +35,12 @@ python_library_host { pkg_path: "edit_monitor", srcs: [ "daemon_manager.py", "edit_monitor.py", ], libs: [ "asuite_cc_client", "edit_event_proto", "watchdog", ], } Loading @@ -53,6 +59,21 @@ python_test_host { }, } python_test_host { name: "edit_monitor_test", main: "edit_monitor_test.py", pkg_path: "edit_monitor", srcs: [ "edit_monitor_test.py", ], libs: [ "edit_monitor_lib", ], test_options: { unit_test: true, }, } python_binary_host { name: "edit_monitor", pkg_path: "edit_monitor", Loading
tools/edit_monitor/edit_monitor.py 0 → 100644 +125 −0 Original line number Diff line number Diff line # Copyright 2024, The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import getpass import logging import multiprocessing.connection import os import platform import time from atest.metrics import clearcut_client from atest.proto import clientanalytics_pb2 from proto import edit_event_pb2 from watchdog.events import FileSystemEvent from watchdog.events import PatternMatchingEventHandler from watchdog.observers import Observer # Enum of the Clearcut log source defined under # /google3/wireless/android/play/playlog/proto/log_source_enum.proto LOG_SOURCE = 2524 class ClearcutEventHandler(PatternMatchingEventHandler): def __init__( self, path: str, cclient: clearcut_client.Clearcut | None = None ): super().__init__(patterns=["*"], ignore_directories=True) self.root_monitoring_path = path self.cclient = cclient or clearcut_client.Clearcut(LOG_SOURCE) self.user_name = getpass.getuser() self.host_name = platform.node() self.source_root = os.environ.get("ANDROID_BUILD_TOP", "") def on_moved(self, event: FileSystemEvent): self._log_edit_event(event, edit_event_pb2.EditEvent.MOVE) def on_created(self, event: FileSystemEvent): self._log_edit_event(event, edit_event_pb2.EditEvent.CREATE) def on_deleted(self, event: FileSystemEvent): self._log_edit_event(event, edit_event_pb2.EditEvent.DELETE) def on_modified(self, event: FileSystemEvent): self._log_edit_event(event, edit_event_pb2.EditEvent.MODIFY) def flushall(self): logging.info("flushing all pending events.") self.cclient.flush_events() def _log_edit_event( self, event: FileSystemEvent, edit_type: edit_event_pb2.EditEvent.EditType ): event_time = time.time() logging.info("%s: %s", event.event_type, event.src_path) try: event_proto = edit_event_pb2.EditEvent( user_name=self.user_name, host_name=self.host_name, source_root=self.source_root, ) event_proto.single_edit_event.CopyFrom( edit_event_pb2.EditEvent.SingleEditEvent( file_path=event.src_path, edit_type=edit_type ) ) clearcut_log_event = clientanalytics_pb2.LogEvent( event_time_ms=int(event_time * 1000), source_extension=event_proto.SerializeToString(), ) self.cclient.log(clearcut_log_event) except Exception: logging.exception("Failed to log edit event.") def start( path: str, cclient: clearcut_client.Clearcut | None = None, pipe_sender: multiprocessing.connection.Connection | None = None, ): """Method to start the edit monitor. This is the entry point to start the edit monitor as a subprocess of the daemon manager. params: path: The root path to monitor cclient: The clearcut client to send the edit logs. conn: the sender of the pipe to communicate with the deamon manager. """ event_handler = ClearcutEventHandler(path, cclient) observer = Observer() logging.info("Starting observer on path %s.", path) observer.schedule(event_handler, path, recursive=True) observer.start() logging.info("Observer started.") if pipe_sender: pipe_sender.send("Observer started.") try: while True: time.sleep(1) finally: event_handler.flushall() observer.stop() observer.join() if pipe_sender: pipe_sender.close()
tools/edit_monitor/edit_monitor_test.py 0 → 100644 +198 −0 Original line number Diff line number Diff line # Copyright 2024, The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Unittests for Edit Monitor.""" import logging import multiprocessing import os import pathlib import signal import sys import tempfile import time import unittest from atest.proto import clientanalytics_pb2 from edit_monitor import edit_monitor from proto import edit_event_pb2 class EditMonitorTest(unittest.TestCase): @classmethod def setUpClass(cls): super().setUpClass() # Configure to print logging to stdout. logging.basicConfig(filename=None, level=logging.DEBUG) console = logging.StreamHandler(sys.stdout) logging.getLogger('').addHandler(console) def setUp(self): super().setUp() self.working_dir = tempfile.TemporaryDirectory() self.root_monitoring_path = pathlib.Path(self.working_dir.name).joinpath( 'files' ) self.root_monitoring_path.mkdir() self.log_event_dir = pathlib.Path(self.working_dir.name).joinpath('logs') self.log_event_dir.mkdir() def tearDown(self): self.working_dir.cleanup() super().tearDown() def test_log_edit_event_success(self): fake_cclient = FakeClearcutClient( log_output_file=self.log_event_dir.joinpath('logs.output') ) p = self._start_test_edit_monitor_process(fake_cclient) # Create and modify a file. test_file = self.root_monitoring_path.joinpath('test.txt') with open(test_file, 'w') as f: f.write('something') # Move the file. test_file_moved = self.root_monitoring_path.joinpath('new_test.txt') test_file.rename(test_file_moved) # Delete the file. test_file_moved.unlink() # Give some time for the edit monitor to receive the edit event. time.sleep(1) # Stop the edit monitor and flush all events. os.kill(p.pid, signal.SIGINT) p.join() logged_events = self._get_logged_events() self.assertEqual(len(logged_events), 4) expected_create_event = edit_event_pb2.EditEvent.SingleEditEvent( file_path=str( self.root_monitoring_path.joinpath('test.txt').resolve() ), edit_type=edit_event_pb2.EditEvent.CREATE, ) expected_modify_event = edit_event_pb2.EditEvent.SingleEditEvent( file_path=str( self.root_monitoring_path.joinpath('test.txt').resolve() ), edit_type=edit_event_pb2.EditEvent.MODIFY, ) expected_move_event = edit_event_pb2.EditEvent.SingleEditEvent( file_path=str( self.root_monitoring_path.joinpath('test.txt').resolve() ), edit_type=edit_event_pb2.EditEvent.MOVE, ) expected_delete_event = edit_event_pb2.EditEvent.SingleEditEvent( file_path=str( self.root_monitoring_path.joinpath('new_test.txt').resolve() ), edit_type=edit_event_pb2.EditEvent.DELETE, ) self.assertEqual( expected_create_event, edit_event_pb2.EditEvent.FromString( logged_events[0].source_extension ).single_edit_event, ) self.assertEqual( expected_modify_event, edit_event_pb2.EditEvent.FromString( logged_events[1].source_extension ).single_edit_event, ) self.assertEqual( expected_move_event, edit_event_pb2.EditEvent.FromString( logged_events[2].source_extension ).single_edit_event, ) self.assertEqual( expected_delete_event, edit_event_pb2.EditEvent.FromString( logged_events[3].source_extension ).single_edit_event, ) def test_log_edit_event_fail(self): fake_cclient = FakeClearcutClient( log_output_file=self.log_event_dir.joinpath('logs.output'), raise_log_exception=True, ) p = self._start_test_edit_monitor_process(fake_cclient) # Create a file. self.root_monitoring_path.joinpath('test.txt').touch() # Give some time for the edit monitor to receive the edit event. time.sleep(1) # Stop the edit monitor and flush all events. os.kill(p.pid, signal.SIGINT) p.join() logged_events = self._get_logged_events() self.assertEqual(len(logged_events), 0) def _start_test_edit_monitor_process( self, cclient ) -> multiprocessing.Process: receiver, sender = multiprocessing.Pipe() # Start edit monitor in a subprocess. p = multiprocessing.Process( target=edit_monitor.start, args=(str(self.root_monitoring_path.resolve()), cclient, sender), ) p.daemon = True p.start() # Wait until observer started. received_data = receiver.recv() self.assertEquals(received_data, 'Observer started.') receiver.close() return p def _get_logged_events(self): with open(self.log_event_dir.joinpath('logs.output'), 'rb') as f: data = f.read() return [ clientanalytics_pb2.LogEvent.FromString(record) for record in data.split(b'\x00') if record ] class FakeClearcutClient: def __init__(self, log_output_file, raise_log_exception=False): self.pending_log_events = [] self.raise_log_exception = raise_log_exception self.log_output_file = log_output_file def log(self, log_event): if self.raise_log_exception: raise Exception('unknown exception') self.pending_log_events.append(log_event) def flush_events(self): delimiter = b'\x00' # Use a null byte as the delimiter with open(self.log_output_file, 'wb') as f: for log_event in self.pending_log_events: f.write(log_event.SerializeToString() + delimiter) self.pending_log_events.clear() if __name__ == '__main__': unittest.main()
