You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@aurora.apache.org by wi...@apache.org on 2015/03/28 01:12:55 UTC
aurora git commit: Kill inotify disk collector.
Repository: aurora
Updated Branches:
refs/heads/master f0a3f9c4f -> 3ab8a9e73
Kill inotify disk collector.
This is currently unwired and its only test is marked flaky.
Reviewed at https://reviews.apache.org/r/32515/
Project: http://git-wip-us.apache.org/repos/asf/aurora/repo
Commit: http://git-wip-us.apache.org/repos/asf/aurora/commit/3ab8a9e7
Tree: http://git-wip-us.apache.org/repos/asf/aurora/tree/3ab8a9e7
Diff: http://git-wip-us.apache.org/repos/asf/aurora/diff/3ab8a9e7
Branch: refs/heads/master
Commit: 3ab8a9e7336701040d9dfa3e482fd13359f6f7db
Parents: f0a3f9c
Author: Brian Wickman <wi...@apache.org>
Authored: Fri Mar 27 17:12:50 2015 -0700
Committer: Brian Wickman <wi...@apache.org>
Committed: Fri Mar 27 17:12:50 2015 -0700
----------------------------------------------------------------------
3rdparty/python/requirements.txt | 2 -
src/main/python/apache/thermos/monitoring/BUILD | 2 -
.../python/apache/thermos/monitoring/disk.py | 116 +------------------
src/test/python/apache/thermos/monitoring/BUILD | 1 +
.../apache/thermos/monitoring/test_disk.py | 20 +---
5 files changed, 4 insertions(+), 137 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/aurora/blob/3ab8a9e7/3rdparty/python/requirements.txt
----------------------------------------------------------------------
diff --git a/3rdparty/python/requirements.txt b/3rdparty/python/requirements.txt
index ee30136..11a307c 100644
--- a/3rdparty/python/requirements.txt
+++ b/3rdparty/python/requirements.txt
@@ -9,7 +9,6 @@ pex==0.8.2
protobuf==2.6.1
psutil==2.1.3
pystachio==0.7.4
-pyyaml==3.10
requests==2.3.0
thrift==0.9.1
twitter.common.app==0.3.3
@@ -31,4 +30,3 @@ twitter.common.rpc==0.3.3
twitter.common.string==0.3.3
twitter.common.testing==0.3.3
twitter.common.zookeeper==0.3.3
-watchdog==0.7.0
http://git-wip-us.apache.org/repos/asf/aurora/blob/3ab8a9e7/src/main/python/apache/thermos/monitoring/BUILD
----------------------------------------------------------------------
diff --git a/src/main/python/apache/thermos/monitoring/BUILD b/src/main/python/apache/thermos/monitoring/BUILD
index a7864d4..633dd95 100644
--- a/src/main/python/apache/thermos/monitoring/BUILD
+++ b/src/main/python/apache/thermos/monitoring/BUILD
@@ -52,12 +52,10 @@ python_library(
name = 'disk',
sources = ['disk.py'],
dependencies = [
- '3rdparty/python:watchdog',
'3rdparty/python:twitter.common.dirutil',
'3rdparty/python:twitter.common.exceptions',
'3rdparty/python:twitter.common.lang',
'3rdparty/python:twitter.common.log',
- '3rdparty/python:twitter.common.quantity',
]
)
http://git-wip-us.apache.org/repos/asf/aurora/blob/3ab8a9e7/src/main/python/apache/thermos/monitoring/disk.py
----------------------------------------------------------------------
diff --git a/src/main/python/apache/thermos/monitoring/disk.py b/src/main/python/apache/thermos/monitoring/disk.py
index ccf6d6a..52c5d74 100644
--- a/src/main/python/apache/thermos/monitoring/disk.py
+++ b/src/main/python/apache/thermos/monitoring/disk.py
@@ -16,33 +16,15 @@
This module provides threads which can be used to gather information on the disk utilisation
under a particular path.
-
-Currently, there are two threads available:
- - DiskCollectorThread, which periodically uses a basic brute-force approach (os.stat()ing every
- file within the path)
- - InotifyDiskCollectorThread, which updates disk utilisation dynamically by using inotify to
- monitor disk changes within the path
-
"""
-import os
import threading
import time
-from Queue import Queue
from twitter.common import log
-from twitter.common.dirutil import du, safe_bsize
+from twitter.common.dirutil import du
from twitter.common.exceptions import ExceptionalThread
from twitter.common.lang import Lockable
-from twitter.common.quantity import Amount, Time
-from watchdog.events import (
- FileCreatedEvent,
- FileDeletedEvent,
- FileModifiedEvent,
- FileMovedEvent,
- FileSystemEventHandler
-)
-from watchdog.observers import Observer as WatchdogObserver
class DiskCollectorThread(ExceptionalThread):
@@ -100,99 +82,3 @@ class DiskCollector(Lockable):
return self._thread.event
else:
return threading.Event()
-
-
-class InotifyDiskCollectorThread(ExceptionalThread, FileSystemEventHandler):
- """ Thread to calculate aggregate disk usage under a given path
-
- Note that while this thread uses inotify (through the watchdog module) to monitor disk events in
- "real time", the actual processing of events is only performed periodically (configured via
- COLLECTION_INTERVAL)
-
- """
- INTERESTING_EVENTS = (FileCreatedEvent, FileDeletedEvent, FileModifiedEvent, FileMovedEvent)
- COLLECTION_INTERVAL = Amount(5, Time.SECONDS)
-
- def __init__(self, path):
- self._path = path
- self._files = {} # file path => size (bytes)
- self._queue = Queue()
- self._observer = WatchdogObserver()
- super(InotifyDiskCollectorThread, self).__init__()
- self.daemon = True
-
- def dispatch(self, event):
- """ Dispatch all interesting events to the internal queue """
- if isinstance(event, self.INTERESTING_EVENTS):
- self._queue.put(event)
-
- def _initialize(self):
- """ Collect an initial snapshot of the disk usage in the path """
- log.debug("Starting watchdog observer to collect events...")
- self._observer.schedule(self, path=self._path, recursive=True)
- self._observer.start()
- log.debug("Collecting initial disk usage sample...")
- for root, _, files in os.walk(self._path):
- for filename in files:
- f = os.path.join(root, filename)
- self._files[f] = safe_bsize(f)
-
- def _process_events(self):
- """ Deduplicate and process watchdog events, updating the internal file store appropriately """
- file_ops = {}
-
- def remove_file(path):
- self._files.pop(path, None)
- def stat_file(path):
- self._files[path] = safe_bsize(path)
-
- while not self._to_process.empty():
- event = self._to_process.get()
- if isinstance(event, (FileCreatedEvent, FileModifiedEvent)):
- file_ops[event.src_path] = lambda: stat_file(event.src_path)
- elif isinstance(event, FileDeletedEvent):
- file_ops[event.src_path] = lambda: remove_file(event.src_path)
- elif isinstance(event, FileMovedEvent):
- file_ops[event.src_path] = lambda: remove_file(event.src_path)
- file_ops[event.dest_path] = lambda: stat_file(event.dest_path)
-
- for op in file_ops.values():
- op()
-
- def run(self):
- """ Loop indefinitely, periodically processing watchdog/inotify events. """
- self._initialize()
- log.debug("Initialization complete. Moving to handling events.")
- while True:
- next = time.time() + self.COLLECTION_INTERVAL.as_(Time.SECONDS)
- if not self._queue.empty():
- self._to_process, self._queue = self._queue, Queue()
- self._process_events()
- time.sleep(max(0, next - time.time()))
-
- @property
- def value(self):
- return sum(self._files.values())
-
-
-class InotifyDiskCollector(object):
- """ Spawn a background thread to sample disk usage """
-
- def __init__(self, root):
- self._root = root
- self._thread = InotifyDiskCollectorThread(self._root)
-
- def sample(self):
- """ Trigger disk collection loop. """
- if not os.path.exists(self._root):
- log.error('Cannot start monitoring path until it exists')
- elif not self._thread.is_alive():
- self._thread.start()
-
- @property
- def value(self):
- return self._thread.value
-
- @property
- def completed_event(self):
- return threading.Event()
http://git-wip-us.apache.org/repos/asf/aurora/blob/3ab8a9e7/src/test/python/apache/thermos/monitoring/BUILD
----------------------------------------------------------------------
diff --git a/src/test/python/apache/thermos/monitoring/BUILD b/src/test/python/apache/thermos/monitoring/BUILD
index 09deebb..89030d0 100644
--- a/src/test/python/apache/thermos/monitoring/BUILD
+++ b/src/test/python/apache/thermos/monitoring/BUILD
@@ -25,6 +25,7 @@ python_tests(name = 'test_disk',
sources = ['test_disk.py'],
dependencies = [
'src/main/python/apache/thermos/monitoring:disk',
+ '3rdparty/python:twitter.common.quantity',
]
)
http://git-wip-us.apache.org/repos/asf/aurora/blob/3ab8a9e7/src/test/python/apache/thermos/monitoring/test_disk.py
----------------------------------------------------------------------
diff --git a/src/test/python/apache/thermos/monitoring/test_disk.py b/src/test/python/apache/thermos/monitoring/test_disk.py
index e137323..e57467c 100644
--- a/src/test/python/apache/thermos/monitoring/test_disk.py
+++ b/src/test/python/apache/thermos/monitoring/test_disk.py
@@ -13,14 +13,12 @@
#
import os
-import time
from tempfile import mkstemp
-import pytest
from twitter.common.dirutil import safe_mkdtemp
-from twitter.common.quantity import Amount, Data, Time
+from twitter.common.quantity import Amount, Data
-from apache.thermos.monitoring.disk import DiskCollector, InotifyDiskCollector
+from apache.thermos.monitoring.disk import DiskCollector
TEST_AMOUNT_1 = Amount(100, Data.MB)
TEST_AMOUNT_2 = Amount(10, Data.MB)
@@ -64,17 +62,3 @@ def test_du_diskcollector():
collector._thread.event.wait()
_run_collector_tests(collector, target, wait)
-
-
-# Flaky test, c.f. https://issues.apache.org/jira/browse/AURORA-916
-@pytest.mark.skipif('True')
-def test_inotify_diskcollector():
- target = safe_mkdtemp()
- INTERVAL = Amount(50, Time.MILLISECONDS)
- collector = InotifyDiskCollector(target)
- collector._thread.COLLECTION_INTERVAL = INTERVAL
-
- def wait():
- time.sleep((2 * INTERVAL).as_(Time.SECONDS))
-
- _run_collector_tests(collector, target, wait)