Merge lp:~notmyname/swift/stats_system into lp:~hudson-openstack/swift/trunk

[John Dickinson]

work in progress of moving the existing internal (pre-openstack) swift stats system to (openstack) swift

[lp:~notmyname/swift/stats_system]

53. By John Dickinson on 2010-08-14

added access log processing plugin

54. By John Dickinson on 2010-08-14

merged with trunk

55. By John Dickinson on 2010-08-17

merged with trunk

56. By John Dickinson on 2010-08-22

merged with trunk

57. By John Dickinson on 2010-08-24

merged with trunk

58. By John Dickinson on 2010-08-26

merged with trunk

59. By John Dickinson on 2010-08-30

merged with trunk

60. By John Dickinson on 2010-09-01

initial tests for the stats system

61. By John Dickinson on 2010-09-01

first test working

62. By John Dickinson on 2010-09-01

merged with trunk

63. By John Dickinson on 2010-09-01

access log parsing tests pass

64. By John Dickinson on 2010-09-01

added (working) stats tests

65. By John Dickinson on 2010-09-02

merged with trunk (utils fix)

66. By John Dickinson on 2010-09-07

updated stats binaries to be DRY compliant

67. By John Dickinson on 2010-09-07

merged with trunk

68. By John Dickinson on 2010-09-10

set up log-stats-collector as a daemon process to create csv files

69. By John Dickinson on 2010-09-10

merged with trunk

70. By John Dickinson on 2010-09-11

added execute perms to stats processor binaries

71. By John Dickinson on 2010-09-13

updated config file loading to work with paste.deploy configs

72. By John Dickinson on 2010-09-13

fixed typos

73. By John Dickinson on 2010-09-14

fixed internal proxy loading

74. By John Dickinson on 2010-09-14

made a memcache stub for the internal proxy server

75. By John Dickinson on 2010-09-15

fixed internal proxy put_container reference

76. By John Dickinson on 2010-09-15

fixed some log uploading glob patterns

77. By John Dickinson on 2010-09-15

fixed bug in calculating offsets for filename patterns

78. By John Dickinson on 2010-09-15

fixed typos in log processor

79. By John Dickinson on 2010-09-15

fixed get_data_list in log_processor

80. By John Dickinson on 2010-09-15

added some debug output

81. By John Dickinson on 2010-09-15

fixed logging and log uploading

82. By John Dickinson on 2010-09-15

fixed copy/paste errors and missing imports

83. By John Dickinson on 2010-09-15

added error handling and missing return statement

84. By John Dickinson on 2010-09-15

handled some typos and better handling of missing data in internal proxy

85. By John Dickinson on 2010-09-15

fixed tests, typos, and added error handling

86. By John Dickinson on 2010-09-15

fixed bug in account stats log processing

87. By John Dickinson on 2010-09-15

fixed listing filter in log processing

88. By John Dickinson on 2010-09-15

fixed stdout capturing for generating csv files

89. By John Dickinson on 2010-09-15

fixed replica count reporting error

90. By John Dickinson on 2010-09-15

fixed lookback in log processor

91. By John Dickinson on 2010-09-15

merged with trunk

92. By John Dickinson on 2010-09-20

merged with trunk

93. By John Dickinson on 2010-09-20

fixed tests to account for changed key name

94. By John Dickinson on 2010-09-20

fixed test bug

95. By John Dickinson on 2010-09-22

updated with changes and suggestions from code review

96. By John Dickinson on 2010-09-22

merged with changes from trunk

97. By John Dickinson on 2010-09-22

added stats overview

98. By John Dickinson on 2010-09-23

updated with changes from trunk

99. By John Dickinson on 2010-09-29

added additional docs

100. By John Dickinson on 2010-09-30

documentation clarification and pep8 fixes

101. By John Dickinson on 2010-09-30

added overview stats to the doc index

102. By John Dickinson on 2010-09-30

made long lines wrap (grr pep8)

103. By John Dickinson on 2010-10-01

merged with trunk

104. By John Dickinson on 2010-10-04

merged with trunk

105. By John Dickinson on 2010-10-04

merged with trunk

106. By John Dickinson on 2010-10-04

fixed stats docs

107. By John Dickinson on 2010-10-04

added a bad lines check to the access log parser

108. By John Dickinson on 2010-10-04

added tests for compressing file reader

109. By John Dickinson on 2010-10-04

fixed compressing file reader test

110. By John Dickinson on 2010-10-04

fixed compressing file reader test

111. By John Dickinson on 2010-10-04

improved compressing file reader test

112. By John Dickinson on 2010-10-04

fixed compressing file reader test

113. By John Dickinson on 2010-10-05

made try/except much less inclusive in access log processor

114. By John Dickinson on 2010-10-05

added keylist mapping tests and fixed other tests

115. By John Dickinson on 2010-10-06

updated stats system tests

116. By John Dickinson on 2010-10-07

merged with gholt's test stubs

117. By John Dickinson on 2010-10-07

updated SAIO instructions for the stats system

118. By John Dickinson on 2010-10-07

fixed stats system saio docs

119. By John Dickinson on 2010-10-07

fixed stats system saio docs

120. By John Dickinson on 2010-10-08

added openstack copyright/license to test_log_processor.py

121. By John Dickinson on 2010-10-08

improved logging in log processors

122. By John Dickinson on 2010-10-08

merged with trunk

123. By John Dickinson on 2010-10-08

fixed missing working directory bug in account stats

124. By John Dickinson on 2010-10-08

fixed readconf parameter that was broken with a previous merge

125. By John Dickinson on 2010-10-08

updated setup.py and saio docs for syats system

126. By John Dickinson on 2010-10-08

added readconf unit test

127. By John Dickinson on 2010-10-08

updated stats saio docs to create logs with the appropriate permissions

128. By John Dickinson on 2010-10-08

fixed bug in log processor internal proxy lazy load code

129. By John Dickinson on 2010-10-08

updated readconf test

130. By John Dickinson on 2010-10-08

updated readconf test

131. By John Dickinson on 2010-10-08

updated readconf test

132. By John Dickinson on 2010-10-08

fixed internal proxy references in log processor

133. By John Dickinson on 2010-10-08

fixed account stats filename creation

134. By John Dickinson on 2010-10-08

pep8 tomfoolery

135. By John Dickinson on 2010-10-08

moved paren

136. By John Dickinson on 2010-10-08

added lazy load of internal proxy to log processor (you were right clay)

Unmerged revisions

=== added file 'bin/swift-account-stats-logger.py'
--- bin/swift-account-stats-logger.py	1970-01-01 00:00:00 +0000
+++ bin/swift-account-stats-logger.py	2010-08-14 18:41:02 +0000
@@ -0,0 +1,81 @@
+#!/usr/bin/python
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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 os
+import signal
+import sys
+import time
+from ConfigParser import ConfigParser
+
+from swift.account_stats import AccountStat
+from swift.common import utils
+
+if __name__ == '__main__':
+    if len(sys.argv) < 2:
+        print "Usage: swift-account-stats-logger CONFIG_FILE"
+        sys.exit()
+
+    c = ConfigParser()
+    if not c.read(sys.argv[1]):
+        print "Unable to read config file."
+        sys.exit(1)
+
+    if c.has_section('log-processor-stats'):
+        stats_conf = dict(c.items('log-processor-stats'))
+    else:
+        print "Unable to find log-processor-stats config section in %s." % \
+                sys.argv[1]
+        sys.exit(1)
+
+    # reference this from the account stats conf
+
+    target_dir = stats.conf.get('log_dir', '/var/log/swift')
+    account_server_conf_loc = stats_conf.get('account_server_conf',
+                                             '/etc/swift/account-server.conf')
+    filename_format = stats.conf['source_filename_format']
+    try:
+        c = ConfigParser()
+        c.read(account_server_conf_loc)
+        account_server_conf = dict(c.items('account-server'))
+    except:
+        print "Unable to load account server conf from %s" % account_server_conf_loc
+        sys.exit(1)
+
+    utils.drop_privileges(account_server_conf.get('user', 'swift'))
+
+    try:
+        os.setsid()
+    except OSError:
+        pass
+
+    logger = utils.get_logger(stats_conf, 'swift-account-stats-logger')
+
+    def kill_children(*args):
+        signal.signal(signal.SIGTERM, signal.SIG_IGN)
+        os.killpg(0, signal.SIGTERM)
+        sys.exit()
+
+    signal.signal(signal.SIGTERM, kill_children)
+
+    stats = AccountStat(filename_format,
+                        target_dir,
+                        account_server_conf,
+                        logger)
+    logger.info("Gathering account stats")
+    start = time.time()
+    stats.find_and_process()
+    logger.info("Gathering account stats complete (%0.2f minutes)" %
+        ((time.time()-start)/60))

=== added file 'bin/swift-log-uploader'
--- bin/swift-log-uploader	1970-01-01 00:00:00 +0000
+++ bin/swift-log-uploader	2010-08-14 18:41:02 +0000
@@ -0,0 +1,83 @@
+#!/usr/bin/python
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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 os
+import signal
+import sys
+import time
+from ConfigParser import ConfigParser
+
+from swift.stats.log_uploader import LogUploader
+from swift.common.utils import get_logger
+
+if __name__ == '__main__':
+    if len(sys.argv) < 3:
+        print "Usage: swift-log-uploader CONFIG_FILE plugin"
+        sys.exit()
+
+    c = ConfigParser()
+    if not c.read(sys.argv[1]):
+        print "Unable to read config file."
+        sys.exit(1)
+
+    if c.has_section('log-processor'):
+        parser_conf = dict(c.items('log-processor'))
+    else:
+        print "Unable to find log-processor config section in %s." % sys.argv[1]
+        sys.exit(1)
+    
+    plugin = sys.argv[2]
+    section_name = 'log-processor-%s' % plugin
+    if c.has_section(section_name):
+        uploader_conf.update(dict(c.items(section_name)))
+    else:
+        print "Unable to find %s config section in %s." % (section_name,
+                                                           sys.argv[1])
+        sys.exit(1)
+
+    try:
+        os.setsid()
+    except OSError:
+        pass
+
+    logger = get_logger(uploader_conf, 'swift-log-uploader')
+
+    def kill_children(*args):
+        signal.signal(signal.SIGTERM, signal.SIG_IGN)
+        os.killpg(0, signal.SIGTERM)
+        sys.exit()
+
+    signal.signal(signal.SIGTERM, kill_children)
+
+    log_dir = uploader_conf.get('log_dir', '/var/log/swift/')
+    swift_account = uploader_conf['swift_account']
+    container_name = uploader_conf['container_name']
+    source_filename_format = uploader_conf['source_filename_format']
+    proxy_server_conf_loc = uploader_conf.get('proxy_server_conf',
+                                              '/etc/swift/proxy-server.conf')
+    try:
+        c = ConfigParser()
+        c.read(proxy_server_conf_loc)
+        proxy_server_conf = dict(c.items('proxy-server'))
+    except:
+        proxy_server_conf = None
+    uploader = LogUploader(log_dir, swift_account, container_name,
+        source_filename_format, proxy_server_conf, logger)
+    logger.info("Uploading logs")
+    start = time.time()
+    uploader.upload_all_logs()
+    logger.info("Uploading logs complete (%0.2f minutes)" %
+        ((time.time()-start)/60))

=== added file 'etc/log-processing.conf-sample'
--- etc/log-processing.conf-sample	1970-01-01 00:00:00 +0000
+++ etc/log-processing.conf-sample	2010-08-14 18:41:02 +0000
@@ -0,0 +1,28 @@
+# plugin section format is named "log-processor-<plugin>"
+# section "log-processor" is the generic defaults (overridden by plugins)
+
+[log-processor]
+# working_dir = /tmp/swift/
+# proxy_server_conf = /etc/swift/proxy-server.conf
+# log_facility = LOG_LOCAL0
+# log_level = INFO
+# lookback_hours = 120
+# lookback_window = 120
+
+[log-processor-access]
+# log_dir = /var/log/swift/
+swift_account = AUTH_7abbc116-8a07-4b63-819d-02715d3e0f31
+container_name = log_data
+source_filename_format = %Y%m%d%H*
+class_path = swift.stats.access_processor.AccessLogProcessor
+# service ips is for client ip addresses that should be counted as servicenet
+# service_ips =
+# server_name = proxy
+
+[log-processor-stats]
+# log_dir = /var/log/swift/
+swift_account = AUTH_7abbc116-8a07-4b63-819d-02715d3e0f31
+container_name = account_stats
+source_filename_format = %Y%m%d%H*
+class_path = swift.stats.stats_processor.StatsLogProcessor
+# account_server_conf = /etc/swift/account-server.conf

=== added file 'swift/common/compressed_file_reader.py'
--- swift/common/compressed_file_reader.py	1970-01-01 00:00:00 +0000
+++ swift/common/compressed_file_reader.py	2010-08-14 18:41:02 +0000
@@ -0,0 +1,72 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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 zlib
+import struct
+
+
+class CompressedFileReader(object):
+    '''
+    Wraps a file object and provides a read method that returns gzip'd data.
+
+    One warning: if read is called with a small value, the data returned may
+    be bigger than the value. In this case, the "compressed" data will be
+    bigger than the original data. To solve this, use a bigger read buffer.
+
+    An example use case:
+    Given an uncompressed file on disk, provide a way to read compressed data
+    without buffering the entire file data in memory. Using this class, an
+    uncompressed log file could be uploaded as compressed data with chunked
+    transfer encoding.
+
+    gzip header and footer code taken from the python stdlib gzip module
+
+    :param file_obj: File object to read from
+    :param compresslevel: compression level
+    '''
+    def __init__(self, file_obj, compresslevel=9):
+        self._f = file_obj
+        self._compressor = zlib.compressobj(compresslevel,
+                                            zlib.DEFLATED,
+                                            -zlib.MAX_WBITS,
+                                            zlib.DEF_MEM_LEVEL,
+                                            0)
+        self.done = False
+        self.first = True
+        self.crc32 = 0
+        self.total_size = 0
+
+    def read(self, *a, **kw):
+        if self.done:
+            return ''
+        x = self._f.read(*a, **kw)
+        if x:
+            self.crc32 = zlib.crc32(x, self.crc32) & 0xffffffffL
+            self.total_size += len(x)
+            compressed = self._compressor.compress(x)
+            if not compressed:
+                compressed = self._compressor.flush(zlib.Z_SYNC_FLUSH)
+        else:
+            compressed = self._compressor.flush(zlib.Z_FINISH)
+            crc32 = struct.pack("<L", self.crc32 & 0xffffffffL)
+            size = struct.pack("<L", self.total_size & 0xffffffffL)
+            footer = crc32 + size
+            compressed += footer
+            self.done = True
+        if self.first:
+            self.first = False
+            header = '\037\213\010\000\000\000\000\000\002\377'
+            compressed = header + compressed
+        return compressed

=== added file 'swift/common/internal_proxy.py'
--- swift/common/internal_proxy.py	1970-01-01 00:00:00 +0000
+++ swift/common/internal_proxy.py	2010-08-14 18:41:02 +0000
@@ -0,0 +1,174 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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 webob
+from urllib import quote, unquote
+from json import loads as json_loads
+
+from swift.common.compressed_file_reader import CompressedFileReader
+from swift.proxy.server import BaseApplication
+
+
+class InternalProxy(object):
+    """
+    Set up a private instance of a proxy server that allows normal requests
+    to be made without having to actually send the request to the proxy.
+    This also doesn't log the requests to the normal proxy logs.
+
+    :param proxy_server_conf: proxy server configuration dictionary
+    :param logger: logger to log requests to
+    :param retries: number of times to retry each request
+    """
+    def __init__(self, proxy_server_conf=None, logger=None, retries=0):
+        self.upload_app = BaseApplication(proxy_server_conf, logger)
+        self.retries = retries
+
+    def upload_file(self, source_file, account, container, object_name,
+                    compress=True, content_type='application/x-gzip'):
+        """
+        Upload a file to cloud files.
+
+        :param source_file: path to or file like object to upload
+        :param account: account to upload to
+        :param container: container to upload to
+        :param object_name: name of object being uploaded
+        :param compress: if True, compresses object as it is uploaded
+        :param content_type: content-type of object
+        :returns: True if successful, False otherwise
+        """
+        log_create_pattern = '/v1/%s/%s/%s' % (account, container, object_name)
+
+        # create the container
+        if not self.put_container(account, container):
+            return False
+
+        # upload the file to the account
+        req = webob.Request.blank(log_create_pattern,
+                            environ={'REQUEST_METHOD': 'PUT'},
+                            headers={'Transfer-Encoding': 'chunked'})
+        if compress:
+            if hasattr(source_file, 'read'):
+                compressed_file = CompressedFileReader(source_file)
+            else:
+                compressed_file = CompressedFileReader(open(source_file, 'rb'))
+            req.body_file = compressed_file
+        else:
+            if not hasattr(source_file, 'read'):
+                source_file = open(source_file, 'rb')
+            req.body_file = source_file
+        req.account = account
+        req.content_type = content_type
+        req.content_length = None   # to make sure we send chunked data
+        resp = self.upload_app.handle_request(self.upload_app.update_request(req))
+        tries = 1
+        while (resp.status_int < 200 or resp.status_int > 299) \
+                and tries <= self.retries:
+            resp = self.upload_app.handle_request(self.upload_app.update_request(req))
+            tries += 1
+        if not (200 <= resp.status_int < 300):
+            return False
+        return True
+
+    def get_object(self, account, container, object_name):
+        """
+        Get object.
+
+        :param account: account name object is in
+        :param container: container name object is in
+        :param object_name: name of object to get
+        :returns: iterator for object data
+        """
+        req = webob.Request.blank('/v1/%s/%s/%s' %
+                            (account, container, object_name),
+                            environ={'REQUEST_METHOD': 'GET'})
+        req.account = account
+        resp = self.upload_app.handle_request(self.upload_app.update_request(req))
+        tries = 1
+        while (resp.status_int < 200 or resp.status_int > 299) \
+                and tries <= self.retries:
+            resp = self.upload_app.handle_request(self.upload_app.update_request(req))
+            tries += 1
+        for x in resp.app_iter:
+            yield x
+
+    def create_container(self, account, container):
+        """
+        Create container.
+
+        :param account: account name to put the container in
+        :param container: container name to create
+        :returns: True if successful, otherwise False
+        """
+        req = webob.Request.blank('/v1/%s/%s' % (account, container),
+                            environ={'REQUEST_METHOD': 'PUT'})
+        req.account = account
+        resp = self.upload_app.handle_request(self.upload_app.update_request(req))
+        tries = 1
+        while (resp.status_int < 200 or resp.status_int > 299) \
+                and tries <= self.retries:
+            resp = self.upload_app.handle_request(self.upload_app.update_request(req))
+            tries += 1
+        return 200 <= resp.status_int < 300
+
+    def get_container_list(self, account, container, marker=None, limit=None,
+                           prefix=None, delimiter=None, full_listing=True):
+        """
+        Get container listing.
+
+        :param account: account name for the container
+        :param container: container name to get the listing of
+        :param marker: marker query
+        :param limit: limit to query
+        :param prefix: prefix query
+        :param delimeter: delimeter for query
+        :param full_listing: if True, make enough requests to get all listings
+        :returns: list of objects
+        """
+        if full_listing:
+            rv = []
+            listing = self.get_container_list(account, container, marker,
+                limit, prefix, delimiter, full_listing=False)
+            while listing:
+                rv.extend(listing)
+                if not delimiter:
+                    marker = listing[-1]['name']
+                else:
+                    marker = listing[-1].get('name', listing[-1].get('subdir'))
+                listing = self.get_container_list(account, container, marker,
+                    limit, prefix, delimiter, full_listing=False)
+            return rv
+        path = '/v1/%s/%s' % (account, container)
+        qs = 'format=json'
+        if marker:
+            qs += '&marker=%s' % quote(marker)
+        if limit:
+            qs += '&limit=%d' % limit
+        if prefix:
+            qs += '&prefix=%s' % quote(prefix)
+        if delimiter:
+            qs += '&delimiter=%s' % quote(delimiter)
+        path += '?%s' % qs
+        req = webob.Request.blank(path, environ={'REQUEST_METHOD': 'GET'})
+        req.account = account
+        resp = self.upload_app.handle_request(self.upload_app.update_request(req))
+        tries = 1
+        while (resp.status_int < 200 or resp.status_int > 299) \
+                and tries <= self.retries:
+            resp = self.upload_app.handle_request(self.upload_app.update_request(req))
+            tries += 1
+        if resp.status_int == 204:
+            return []
+        if 200 <= resp.status_int < 300:
+            return json_loads(resp.body)

=== added directory 'swift/stats'
=== added file 'swift/stats/__init__.py'
=== added file 'swift/stats/access_processor.py'
--- swift/stats/access_processor.py	1970-01-01 00:00:00 +0000
+++ swift/stats/access_processor.py	2010-08-14 18:41:02 +0000
@@ -0,0 +1,168 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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.
+
+class AccessLogProcessor(object):
+
+    def __init__(self, conf):
+        self.server_name = conf.get('server_name', 'proxy')
+
+    def _log_line_parser(self, raw_log):
+        '''given a raw access log line, return a dict of the good parts'''
+        d = {}
+        try:
+            (_,
+            server,
+            client_ip,
+            lb_ip,
+            timestamp,
+            method,
+            request,
+            http_version,
+            code,
+            referrer,
+            user_agent,
+            auth_token,
+            bytes_in,
+            bytes_out,
+            etag,
+            trans_id,
+            headers,
+            processing_time) = (unquote(x) for x in raw_log[16:].split(' '))
+            if server != self.server_name:
+                raise ValueError('incorrect server name in log line')
+            (version,
+            account,
+            container_name,
+            object_name) = split_path(request, 2, 4, True)
+            if container_name is not None:
+                container_name = container_name.split('?', 1)[0]
+            if object_name is not None:
+                object_name = object_name.split('?', 1)[0]
+            account = account.split('?', 1)[0]
+            query = None
+            if '?' in request:
+                request, query = request.split('?', 1)
+                args = query.split('&')
+                # Count each query argument. This is used later to aggregate
+                # the number of format, prefix, etc. queries.
+                for q in args:
+                    if '=' in q:
+                        k, v = q.split('=', 1)
+                    else:
+                        k = q
+                    # Certain keys will get summmed in stats reporting
+                    # (format, path, delimiter, etc.). Save a "1" here
+                    # to indicate that this request is 1 request for
+                    # its respective key.
+                    d[k] = 1
+        except ValueError:
+            pass
+        else:
+            d['client_ip'] = client_ip
+            d['lb_ip'] = lb_ip
+            d['method'] = method
+            d['request'] = request
+            if query:
+                d['query'] = query
+            d['http_version'] = http_version
+            d['code'] = code
+            d['referrer'] = referrer
+            d['user_agent'] = user_agent
+            d['auth_token'] = auth_token
+            d['bytes_in'] = bytes_in
+            d['bytes_out'] = bytes_out
+            d['etag'] = etag
+            d['trans_id'] = trans_id
+            d['processing_time'] = processing_time
+            day, month, year, hour, minute, second = timestamp.split('/')
+            d['day'] = day
+            month = ('%02s' % month_map.index(month)).replace(' ', '0')
+            d['month'] = month
+            d['year'] = year
+            d['hour'] = hour
+            d['minute'] = minute
+            d['second'] = second
+            d['tz'] = '+0000'
+            d['account'] = account
+            d['container_name'] = container_name
+            d['object_name'] = object_name
+            d['bytes_out'] = int(d['bytes_out'].replace('-','0'))
+            d['bytes_in'] = int(d['bytes_in'].replace('-','0'))
+            d['code'] = int(d['code'])
+        return d
+
+    def process(self, obj_stream):
+        '''generate hourly groupings of data from one access log file'''
+        hourly_aggr_info = {}
+        aggr_account_logs = {}
+        container_line_counts = collections.defaultdict(int)
+        log_buffer = collections.defaultdict(list)
+        for line in obj_stream:
+            line_data = self._log_line_parser(line)
+            if not line_data:
+                continue
+            account = line_data['account']
+            container_name = line_data['container_name']
+            year = line_data['year']
+            month = line_data['month']
+            day = line_data['day']
+            hour = line_data['hour']
+            bytes_out = line_data['bytes_out']
+            bytes_in = line_data['bytes_in']
+            method = line_data['method']
+            code = int(line_data['code'])
+            object_name = line_data['object_name']
+            client_ip = line_data['client_ip']
+
+            op_level = None
+            if not container_name:
+                op_level = 'account'
+            elif container_name and not object_name:
+                op_level = 'container'
+            elif object_name:
+                op_level = 'object'
+
+            aggr_key = (account, year, month, day, hour)
+            d = hourly_aggr_info.get(aggr_key, {})
+            if line_data['lb_ip'] in self.lb_private_ips:
+                source = 'service'
+            else:
+                source = 'public'
+            
+            if line_data['client_ip'] in self.service_ips:
+                source = 'service'
+
+            d[(source, 'bytes_out')] = d.setdefault((source, 'bytes_out'), 0) + \
+                                       bytes_out
+            d[(source, 'bytes_in')] = d.setdefault((source, 'bytes_in'), 0) + \
+                                      bytes_in
+
+            d['format_query'] = d.setdefault('format_query', 0) + \
+                                line_data.get('format', 0)
+            d['marker_query'] = d.setdefault('marker_query', 0) + \
+                                line_data.get('marker', 0)
+            d['prefix_query'] = d.setdefault('prefix_query', 0) + \
+                                line_data.get('prefix', 0)
+            d['delimiter_query'] = d.setdefault('delimiter_query', 0) + \
+                                   line_data.get('delimiter', 0)
+            path = line_data.get('path', 0)
+            d['path_query'] = d.setdefault('path_query', 0) + path
+
+            code = '%dxx' % (code/100)
+            key = (source, op_level, method, code)
+            d[key] = d.setdefault(key, 0) + 1
+
+            hourly_aggr_info[aggr_key] = d
+        return hourly_aggr_info, item, aggr_account_logs

=== added file 'swift/stats/account_stats.py'
--- swift/stats/account_stats.py	1970-01-01 00:00:00 +0000
+++ swift/stats/account_stats.py	2010-08-14 18:41:02 +0000
@@ -0,0 +1,69 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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 os
+import time
+
+from swift.account.server import DATADIR as account_server_data_dir
+from swift.common.db import AccountBroker
+from swift.common.internal_proxy import InternalProxy
+from swift.common.utils import renamer
+
+class AccountStat(object):
+    def __init__(self, filename_format, target_dir, server_conf, logger):
+        self.filename_format = filename_format
+        self.target_dir = target_dir
+        self.devices = server_conf.get('devices', '/srv/node')
+        self.mount_check = server_conf.get('mount_check', 'true').lower() in \
+                              ('true', 't', '1', 'on', 'yes', 'y')
+        self.logger = logger
+
+    def find_and_process(self):
+        src_filename = time.strftime(self.filename_format)
+        tmp_filename = os.path.join('/tmp', src_filename)
+        with open(tmp_filename, 'wb') as statfile:
+            #statfile.write('Account Name, Container Count, Object Count, Bytes Used, Created At\n')
+            for device in os.listdir(self.devices):
+                if self.mount_check and \
+                        not os.path.ismount(os.path.join(self.devices, device)):
+                    self.logger.error("Device %s is not mounted, skipping." %
+                        device)
+                    continue
+                accounts = os.path.join(self.devices,
+                                        device,
+                                        account_server_data_dir)
+                if not os.path.exists(accounts):
+                    self.logger.debug("Path %s does not exist, skipping." %
+                        accounts)
+                    continue
+                for root, dirs, files in os.walk(accounts, topdown=False):
+                    for filename in files:
+                        if filename.endswith('.db'):
+                            broker = AccountBroker(os.path.join(root, filename))
+                            if not broker.is_deleted():
+                                account_name,
+                                created_at,
+                                _, _,
+                                container_count,
+                                object_count,
+                                bytes_used,
+                                _, _ = broker.get_info()
+                                line_data = '"%s",%d,%d,%d,%s\n' % (account_name,
+                                                                    container_count,
+                                                                    object_count,
+                                                                    bytes_used,
+                                                                    created_at)
+                                statfile.write(line_data)
+        renamer(tmp_filename, os.path.join(self.target_dir, src_filename))

=== added file 'swift/stats/log_processor.py'
--- swift/stats/log_processor.py	1970-01-01 00:00:00 +0000
+++ swift/stats/log_processor.py	2010-08-14 18:41:02 +0000
@@ -0,0 +1,226 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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.
+
+class LogProcessor(object):
+
+    def __init__(self, conf, logger):
+        stats_conf = conf.get('log-processor', {})
+        
+        working_dir = stats_conf.get('working_dir', '/tmp/swift/')
+        if working_dir.endswith('/') and len(working_dir) > 1:
+            working_dir = working_dir[:-1]
+        self.working_dir = working_dir
+        proxy_server_conf_loc = stats_conf.get('proxy_server_conf',
+                                               '/etc/swift/proxy-server.conf')
+        try:
+            c = ConfigParser()
+            c.read(proxy_server_conf_loc)
+            proxy_server_conf = dict(c.items('proxy-server'))
+        except:
+            proxy_server_conf = None
+        self.proxy_server_conf = proxy_server_conf
+        if isinstance(logger, tuple):
+            self.logger = get_logger(*logger)
+        else:
+            self.logger = logger
+        
+        # load the processing plugins
+        self.plugins = {}
+        plugin_prefix = 'log-processor-'
+        for section in (x for x in conf if x.startswith(plugin_prefix)):
+            plugin_name = section[len(plugin_prefix):]
+            plugin_conf = conf.get(section, {})
+            self.plugins[plugin_name] = plugin_conf
+            import_target, class_name = plugin_conf['class_path'].rsplit('.', 1)
+            module = __import__(import_target, fromlist=[import_target])
+            klass = getattr(module, class_name)
+            self.plugins[plugin_name]['instance'] = klass(plugin_conf)
+
+    def process_one_file(self, plugin_name, account, container, object_name):
+        # get an iter of the object data
+        compressed = object_name.endswith('.gz')
+        stream = self.get_object_data(account, container, object_name,
+                                      compressed=compressed)
+        # look up the correct plugin and send the stream to it
+        return self.plugins[plugin_name]['instance'].process(stream)
+
+    def get_data_list(self, start_date=None, end_date=None, listing_filter=None):
+        total_list = []
+        for p in self.plugins:
+            account = p['swift_account']
+            container = p['container_name']
+            l = self.get_container_listing(account, container, start_date,
+                                           end_date, listing_filter)
+            for i in l:
+                total_list.append((p, account, container, i))
+        return total_list
+
+    def get_container_listing(self, swift_account, container_name, start_date=None,
+                         end_date=None, listing_filter=None):
+        '''
+        Get a container listing, filtered by start_date, end_date, and
+        listing_filter. Dates, if given, should be in YYYYMMDDHH format
+        '''
+        search_key = None
+        if start_date is not None:
+            date_parts = []
+            try:
+                year, start_date = start_date[:4], start_date[4:]
+                if year:
+                    date_parts.append(year)
+                    month, start_date = start_date[:2], start_date[2:]
+                    if month:
+                        date_parts.append(month)
+                        day, start_date = start_date[:2], start_date[2:]
+                        if day:
+                            date_parts.append(day)
+                            hour, start_date = start_date[:2], start_date[2:]
+                            if hour:
+                                date_parts.append(hour)
+            except IndexError:
+                pass
+            else:
+                search_key = '/'.join(date_parts)
+        end_key = None
+        if end_date is not None:
+            date_parts = []
+            try:
+                year, end_date = end_date[:4], end_date[4:]
+                if year:
+                    date_parts.append(year)
+                    month, end_date = end_date[:2], end_date[2:]
+                    if month:
+                        date_parts.append(month)
+                        day, end_date = end_date[:2], end_date[2:]
+                        if day:
+                            date_parts.append(day)
+                            hour, end_date = end_date[:2], end_date[2:]
+                            if hour:
+                                date_parts.append(hour)
+            except IndexError:
+                pass
+            else:
+                end_key = '/'.join(date_parts)
+        container_listing = self.private_proxy.get_container_list(
+                                    swift_account,
+                                    container_name,
+                                    marker=search_key)
+        results = []
+        if container_listing is not None:
+            if listing_filter is None:
+                listing_filter = set()
+            for item in container_listing:
+                name = item['name']
+                if end_key and name > end_key:
+                    break
+                if name not in listing_filter:
+                    results.append(name)
+        return results
+
+    def get_object_data(self, swift_account, container_name, object_name,
+                        compressed=False):
+        '''reads an object and yields its lines'''
+        o = self.private_proxy.get_object(swift_account,
+                                          container_name,
+                                          object_name)
+        tmp_file = tempfile.TemporaryFile(dir=self.working_dir)
+        with tmp_file as f:
+            bad_file = False
+            try:
+                for chunk in o:
+                    f.write(chunk)
+            except ChunkReadTimeout:
+                bad_file = True
+            if bad_file:
+                raise BadFileDownload()
+            f.flush()
+            f.seek(0) # rewind to start reading
+            last_part = ''
+            last_compressed_part = ''
+            # magic in the following zlib.decompressobj argument is courtesy of
+            # http://stackoverflow.com/questions/2423866/python-decompressing-gzip-chunk-by-chunk
+            d = zlib.decompressobj(16+zlib.MAX_WBITS)
+            for chunk in iter(lambda: f.read(16384), ''):
+                if compressed:
+                    try:
+                        chunk = d.decompress(chunk)
+                    except zlib.error:
+                        raise BadFileDownload() # bad compressed data
+                parts = chunk.split('\n')
+                parts[0] = last_part + parts[0]
+                for part in parts[:-1]:
+                    yield part
+                last_part = parts[-1]
+            if last_part:
+                yield last_part
+
+def multiprocess_collate(processor_args,
+                         start_date=None,
+                         end_date=None,
+                         listing_filter=None):
+    '''get listing of files and yield hourly data from them'''
+    p = LogProcessor(*processor_args)
+    all_files = p.get_data_list(start_date, end_date, listing_filter)
+
+    p.logger.info('loaded %d files to process' % len(all_files))
+
+    if not all_files:
+        # no work to do
+        return
+
+    worker_count = multiprocessing.cpu_count() - 1
+    results = []
+    in_queue = multiprocessing.Queue()
+    out_queue = multiprocessing.Queue()
+    for _ in range(worker_count):
+        p = multiprocessing.Process(target=collate_worker,
+                                    args=(processor_args,
+                                          in_queue,
+                                          out_queue))
+        p.start()
+        results.append(p)
+    for x in all_files:
+        in_queue.put(x)
+    for _ in range(worker_count):
+        in_queue.put(None)
+    count = 0
+    while True:
+        try:
+            item, data = out_queue.get_nowait()
+            count += 1
+            if data:
+                yield item, data
+            if count >= len(all_files):
+                # this implies that one result will come from every request
+                break
+        except Queue.Empty:
+            time.sleep(.1)
+    for r in results:
+        r.join()
+
+def collate_worker(processor_args, in_queue, out_queue):
+    '''worker process for multiprocess_collate'''
+    p = LogProcessor(*processor_args)
+    while True:
+        try:
+            item = in_queue.get_nowait()
+            if item is None:
+                break
+        except Queue.Empty:
+            time.sleep(.1)
+        else:
+            ret = None
+            ret = p.process_one_file(item)
+            out_queue.put((item, ret))
\ No newline at end of file

=== added file 'swift/stats/log_uploader.py'
--- swift/stats/log_uploader.py	1970-01-01 00:00:00 +0000
+++ swift/stats/log_uploader.py	2010-08-14 18:41:02 +0000
@@ -0,0 +1,135 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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.
+
+from __future__ import with_statement
+import os
+import hashlib
+import time
+import gzip
+import glob
+
+from swift.common.internal_proxy import InternalProxy
+
+class LogUploader(object):
+    '''
+    Given a local directory, a swift account, and a container name, LogParser
+    will upload all files in the local directory to the given account/container.
+    All but the newest files will be uploaded, and the files' md5 sum will be
+    computed. The hash is used to prevent duplicate data from being uploaded
+    multiple times in different files (ex: log lines). Since the hash is
+    computed, it is also used as the uploaded object's etag to ensure data
+    integrity.
+    
+    Note that after the file is successfully uploaded, it will be unlinked.
+    
+    The given proxy server config is used to instantiate a proxy server for
+    the object uploads.
+    '''
+
+    def __init__(self, log_dir, swift_account, container_name, filename_format,
+            proxy_server_conf, logger):
+        if not log_dir.endswith('/'):
+            log_dir = log_dir + '/'
+        self.log_dir = log_dir
+        self.swift_account = swift_account
+        self.container_name = container_name
+        self.filename_format = filename_format
+        self.internal_proxy = InternalProxy(proxy_server_conf, logger)
+        self.logger = logger
+
+    def upload_all_logs(self):
+        i = [(c,self.filename_format.index(c)) for c in '%Y %m %d %H'.split()]
+        i.sort()
+        year_offset = month_offset = day_offset = hour_offset = None
+        for c, start in i:
+            if c == '%Y':
+                year_offset = start, start+4
+            elif c == '%m':
+                month_offset = start, start+2
+            elif c == '%d':
+                day_offset = start, start+2
+            elif c == '%H':
+                hour_offset = start, start+2
+        if not (year_offset and month_offset and day_offset and hour_offset):
+            # don't have all the parts, can't upload anything
+            return
+        glob_pattern = self.filename_format
+        glob_pattern = glob_pattern.replace('%Y', '????')
+        glob_pattern = glob_pattern.replace('%m', '??')
+        glob_pattern = glob_pattern.replace('%d', '??')
+        glob_pattern = glob_pattern.replace('%H', '??')
+        filelist = glob.iglob(os.path.join(self.log_dir, glob_pattern))
+        current_hour = int(time.strftime('%H'))
+        today = int(time.strftime('%Y%m%d'))
+        self.internal_proxy.create_container(self.swift_account,
+                                            self.container_name)
+        for filename in filelist:
+            try:
+                # From the filename, we need to derive the year, month, day,
+                # and hour for the file. These values are used in the uploaded
+                # object's name, so they should be a reasonably accurate
+                # representation of the time for which the data in the file was
+                # collected. The file's last modified time is not a reliable
+                # representation of the data in the file. For example, an old
+                # log file (from hour A) may be uploaded or moved into the
+                # log_dir in hour Z. The file's modified time will be for hour
+                # Z, and therefore the object's name in the system will not
+                # represent the data in it.
+                # If the filename doesn't match the format, it shouldn't be
+                # uploaded.
+                year = filename[slice(*year_offset)]
+                month = filename[slice(*month_offset)]
+                day = filename[slice(*day_offset)]
+                hour = filename[slice(*hour_offset)]
+            except IndexError:
+                # unexpected filename format, move on
+                self.logger.error("Unexpected log: %s" % filename)
+                continue
+            if (time.time() - os.stat(filename).st_mtime) < 7200:
+                # don't process very new logs
+                self.logger.debug("Skipping log: %s (< 2 hours old)" % filename)
+                continue
+            self.upload_one_log(filename, year, month, day, hour)
+
+    def upload_one_log(self, filename, year, month, day, hour):
+        if os.path.getsize(filename) == 0:
+            self.logger.debug("Log %s is 0 length, skipping" % filename)
+            return
+        self.logger.debug("Processing log: %s" % filename)
+        filehash = hashlib.md5()
+        already_compressed = True if filename.endswith('.gz') else False
+        opener = gzip.open if already_compressed else open
+        f = opener(filename, 'rb')
+        try:
+            for line in f:
+                # filter out bad lines here?
+                filehash.update(line)
+        finally:
+            f.close()
+        filehash = filehash.hexdigest()
+        # By adding a hash to the filename, we ensure that uploaded files
+        # have unique filenames and protect against uploading one file
+        # more than one time. By using md5, we get an etag for free.
+        target_filename = '/'.join([year, month, day, hour, filehash+'.gz'])
+        if self.internal_proxy.upload_file(filename,
+                                          self.swift_account,
+                                          self.container_name,
+                                          target_filename,
+                                          compress=(not already_compressed)):
+            self.logger.debug("Uploaded log %s to %s" %
+                (filename, target_filename))
+            os.unlink(filename)
+        else:
+            self.logger.error("ERROR: Upload of log %s failed!" % filename)

=== added file 'swift/stats/stats_processor.py'
--- swift/stats/stats_processor.py	1970-01-01 00:00:00 +0000
+++ swift/stats/stats_processor.py	2010-08-14 18:41:02 +0000
@@ -0,0 +1,54 @@
+# Copyright (c) 2010 OpenStack, LLC.
+#
+# 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.
+
+class StatsLogProcessor(object):
+
+    def __init__(self, conf):
+        pass
+
+    def process(self, obj_stream):
+        '''generate hourly groupings of data from one stats log file'''
+        account_totals = {}
+        year, month, day, hour, _ = item.split('/')
+        for line in obj_stream:
+            if not line:
+                continue
+            try:
+                (account,
+                container_count,
+                object_count,
+                bytes_used,
+                created_at) = line.split(',')
+                account = account.strip('"')
+                if account_name and account_name != account:
+                    continue
+                container_count = int(container_count.strip('"'))
+                object_count = int(object_count.strip('"'))
+                bytes_used = int(bytes_used.strip('"'))
+                aggr_key = (account, year, month, day, hour)
+                d = account_totals.get(aggr_key, {})
+                d['count'] = d.setdefault('count', 0) + 1
+                d['container_count'] = d.setdefault('container_count', 0) + \
+                                       container_count
+                d['object_count'] = d.setdefault('object_count', 0) + \
+                                    object_count
+                d['bytes_used'] = d.setdefault('bytes_used', 0) + \
+                                  bytes_used
+                d['created_at'] = created_at
+                account_totals[aggr_key] = d
+            except (IndexError, ValueError):
+                # bad line data
+                pass
+        return account_totals, item