reporting.py

import argparse
import ast
from concurrent.futures import ThreadPoolExecutor
import json
import logging
import re
import redis
import requests
import sys
from threading import Thread
import time
import urllib

logging.basicConfig(level=logging.INFO)
_log = logging.getLogger('utapi-reindex:reporting')

SENTINEL_CONNECT_TIMEOUT_SECONDS = 10
EXIT_CODE_SENTINEL_CONNECTION_ERROR = 100

def get_options():
    parser = argparse.ArgumentParser()
    parser.add_argument("-i", "--sentinel-ip", default='127.0.0.1', help="Sentinel IP")
    parser.add_argument("-p", "--sentinel-port", default="16379", help="Sentinel Port")
    parser.add_argument("-v", "--redis-password", default=None, help="Redis AUTH Password")
    parser.add_argument("-n", "--sentinel-cluster-name", default='scality-s3', help="Redis cluster name")
    parser.add_argument("-b", "--bucketd-addr", default='http://127.0.0.1:9000', help="URL of the bucketd server")
    return parser.parse_args()

def safe_print(content):
    print("{0}".format(content))


class askRedis():

    def __init__(self, ip="127.0.0.1", port="16379", sentinel_cluster_name="scality-s3", password=None):
        self._password = password
        r = redis.Redis(
            host=ip,
            port=port,
            db=0,
            password=password,
            socket_connect_timeout=SENTINEL_CONNECT_TIMEOUT_SECONDS
        )
        try:
            self._ip, self._port = r.sentinel_get_master_addr_by_name(sentinel_cluster_name)
        except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError) as e:
            _log.error(f'Failed to connect to redis sentinel at {ip}:{port}: {e}')
            # use a specific error code to hint on retrying with another sentinel node
            sys.exit(EXIT_CODE_SENTINEL_CONNECTION_ERROR)

    def read(self, resource, name):
        r = redis.Redis(host=self._ip, port=self._port, db=0, password=self._password)
        res = 's3:%s:%s:storageUtilized:counter' % (resource, name)
        total_size = r.get(res)
        res = 's3:%s:%s:numberOfObjects:counter' % (resource, name)
        files = r.get(res)
        try:
            return {'files': int(files), "total_size": int(total_size)}
        except Exception as e:
            return {'files': 0, "total_size": 0}


class S3ListBuckets():

    def __init__(self, host='127.0.0.1:9000'):
        self.bucketd_host = host

    def run(self):
        docs = []
        url = "%s/default/bucket/users..bucket" % self.bucketd_host
        session = requests.Session()
        marker = ''
        page = 0
        while True:
            params = {'marker': marker, 'maxKeys': 1000}
            r = session.get(url, params=params, timeout=30)
            if r.status_code != 200:
                _log.error('users..bucket bucketd listing failed with status: %s', r.status_code)
                break
            payload = json.loads(r.text)
            contents = payload.get('Contents', [])
            page += 1
            _log.info('Listing page %d: got %d entries (marker=%r)', page, len(contents), marker)
            for keys in contents:
                key = keys["key"]
                r1 = re.match("(\w+)..\|..(\w+.*)", key)
                if r1:
                    docs.append(r1.groups())
                else:
                    _log.warning('Skipping unexpected key format: %s', key)
            if not payload.get('IsTruncated') or not contents:
                _log.info('Listing complete: %d pages, %d buckets total', page, len(docs))
                break
            marker = contents[-1]['key']
        return docs

if __name__ == '__main__':
    options = get_options()
    redis_conf = dict(
        ip=options.sentinel_ip,
        port=options.sentinel_port,
        sentinel_cluster_name=options.sentinel_cluster_name,
        password=options.redis_password
    )

    P = S3ListBuckets(options.bucketd_addr)
    listbuckets = P.run()

    userids = set([x for x, y in listbuckets])

    executor = ThreadPoolExecutor(max_workers=1)
    for userid, bucket in listbuckets:
        U = askRedis(**redis_conf)
        data = U.read('buckets', bucket)
        content = "Account:%s|Bucket:%s|NumberOFfiles:%s|StorageCapacity:%s " % (
            userid, bucket, data["files"], data["total_size"])
        executor.submit(safe_print, content)
        data = U.read('buckets', 'mpuShadowBucket'+bucket)
        content = "Account:%s|Bucket:%s|NumberOFfiles:%s|StorageCapacity:%s " % (
            userid, 'mpuShadowBucket'+bucket, data["files"], data["total_size"])
        executor.submit(safe_print, content)


    executor.submit(safe_print, "")
    for userid in sorted(userids):
        U = askRedis(**redis_conf)
        data = U.read('accounts', userid)
        content = "Account:%s|NumberOFfiles:%s|StorageCapacity:%s " % (
            userid, data["files"], data["total_size"])
        executor.submit(safe_print, content)