<commit_before>#
# Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
#

import re
from cfgm_common import jsonutils as json
import time
import gevent
import disc_consts
import disc_exceptions
from datetime import datetime
from gevent.coros import BoundedSemaphore

import pycassa
import pycassa.util
from pycassa.system_manager import *
from pycassa.util import *
from pycassa.types import *
from sandesh_common.vns.constants import DISCOVERY_SERVER_KEYSPACE_NAME, \
    CASSANDRA_DEFAULT_GC_GRACE_SECONDS

class DiscoveryCassandraClient(object):
    _DISCOVERY_KEYSPACE_NAME = DISCOVERY_SERVER_KEYSPACE_NAME
    _DISCOVERY_CF_NAME = 'discovery'

    @classmethod
    def get_db_info(cls):
        db_info = [(cls._DISCOVERY_KEYSPACE_NAME, [cls._DISCOVERY_CF_NAME])]
        return db_info
    # end get_db_info

    def __init__(self, module, cass_srv_list, reset_config=False,
                 max_retries=5, timeout=5, cass_credential=None):
        self._disco_cf_name = 'discovery'
        self._keyspace_name = 'DISCOVERY_SERVER'
        self._reset_config = reset_config
        self._credential = cass_credential
        self._cassandra_init(cass_srv_list, max_retries, timeout)

        self._debug = {
        }
    #end __init__

    # Helper routines for cassandra
    def _cassandra_init(self, server_list, max_retries, timeout):

        # column name <table-name>, <id1>, <id2>
        disco_cf_info = (self._disco_cf_name, 
            CompositeType(AsciiType(), UTF8Type(), UTF8Type()), AsciiType())

        # 1. Ensure keyspace and schema/CFs exist
        self._cassandra_ensure_keyspace(server_list, self._keyspace_name,
                [disco_cf_info])

        pool = pycassa.ConnectionPool(self._keyspace_name,
                                      server_list, max_overflow=-1,
                                      use_threadlocal=True, prefill=True,
                                      pool_size=100, pool_timeout=120,
                                      max_retries=max_retries, timeout=timeout,
                                      credentials=self._credential)
        rd_consistency = pycassa.cassandra.ttypes.ConsistencyLevel.ONE
        wr_consistency = pycassa.cassandra.ttypes.ConsistencyLevel.ONE
        self._disco_cf = pycassa.ColumnFamily(pool, self._disco_cf_name,
                                              read_consistency_level = rd_consistency,
                                              write_consistency_level = wr_consistency)
    #end _cassandra_init

    def _cassandra_ensure_keyspace(self, server_list,
                                   keyspace_name, cf_info_list):
        # Retry till cassandra is up
        server_idx = 0
        num_dbnodes = len(server_list)
        connected = False
        while not connected:
            try:
                cass_server = server_list[server_idx]
                sys_mgr = SystemManager(cass_server,credentials=self._credential)
                connected = True
            except Exception as e:
                # TODO do only for thrift.transport.TTransport.TTransportException
                server_idx = (server_idx + 1) % num_dbnodes
                time.sleep(3)

        if self._reset_config:
            try:
                sys_mgr.drop_keyspace(keyspace_name)
            except pycassa.cassandra.ttypes.InvalidRequestException as e:
                # TODO verify only EEXISTS
                print "Warning! " + str(e)

        try:
            # TODO replication_factor adjust?
            sys_mgr.create_keyspace(keyspace_name, SIMPLE_STRATEGY,
                                    {'replication_factor': str(num_dbnodes)})
        except pycassa.cassandra.ttypes.InvalidRequestException as e:
            # TODO verify only EEXISTS
            print "Warning! " + str(e)

        for cf_info in cf_info_list:
            try:
                (cf_name, comparator_type, validator_type) = cf_info
                sys_mgr.create_column_family(keyspace_name, cf_name, 
                        comparator_type = comparator_type, default_validation_class = validator_type)
                sys_mgr.alter_column_family(keyspace_name, cf_name,
                    gc_grace_seconds=CASSANDRA_DEFAULT_GC_GRACE_SECONDS)
            except pycassa.cassandra.ttypes.InvalidRequestException as e:
                # TODO verify only EEXISTS
                print "Warning! " + str(e)
    #end _cassandra_ensure_keyspace

    def get_debug_stats(self):
        return self._debug
    # end

    """
        various column names
        ('client', client_id, 'client-entry')
        ('subscriber', service_id, client_id)
        ('subscription', client_id, service_id)
        ('service', service_id, 'service-entry')
    """

    # decorator to catch connectivity error
    def cass_error_handler(func):
        def error_handler(*args, **kwargs):
            try:
                return func(*args,**kwargs)
            except (pycassa.pool.AllServersUnavailable,
                    pycassa.pool.MaximumRetryException):
                raise disc_exceptions.ServiceUnavailable()
            except Exception as e:
                raise
        return error_handler

    # return all publisher entries
    @cass_error_handler
    def service_entries(self, service_type = None):
        col_name = ('service',)
        try:
            data = self._disco_cf.get_range(column_start = col_name, column_finish = col_name)
            for service_type, services in data:
                for col_name in services:
                    col_value = services[col_name]
                    entry = json.loads(col_value)
                    col_name = ('subscriber', entry['service_id'],)
                    entry['in_use'] = self._disco_cf.get_count(service_type, 
                        column_start = col_name, column_finish = col_name)
                    yield(entry)
        except pycassa.pool.AllServersUnavailable:
            raise disc_exceptions.ServiceUnavailable()
            #raise StopIteration

    # return all clients
    def subscriber_entries(self):
        col_name = ('client',)
        data = self._disco_cf.get_range(column_start = col_name, column_finish = col_name)
        for service_type, clients in data:
            for col_name in clients:
                (foo, client_id, service_id) = col_name
                # skip pure client entry
                if service_id == disc_consts.CLIENT_TAG:
                    continue
                yield((client_id, service_type))
    # end

    # return all subscriptions
    @cass_error_handler
    def get_all_clients(self, service_type=None, service_id=None):
        r = []
        entry_format_subscriber = False
        if service_type and service_id:
            # ('subscriber', service_id, client_id)
            col_name = ('subscriber', service_id,)
            try:
                clients = self._disco_cf.get(service_type, column_start = col_name,
                    column_finish = col_name, column_count = disc_consts.MAX_COL)
            except pycassa.NotFoundException:
                return None
            data = [(service_type, dict(clients))]
            entry_format_subscriber = True
        elif service_type:
            col_name = ('client', )
            try:
                clients = self._disco_cf.get(service_type, column_start = col_name,
                    column_finish = col_name, column_count = disc_consts.MAX_COL)
            except pycassa.NotFoundException:
                return None
            data = [(service_type, dict(clients))]
        else:
            col_name = ('client', )
            try:
                data = self._disco_cf.get_range(column_start=col_name, column_finish=col_name)
            except pycassa.NotFoundException:
                return None

        for service_type, clients in data:
            rr = []
            for col_name in clients:
                if entry_format_subscriber:
                    (foo, service_id, client_id) = col_name
                else:
                    (foo, client_id, service_id) = col_name
                    # skip pure client entry
                    if service_id == disc_consts.CLIENT_TAG:
                        continue
                entry_str = clients[col_name]
                entry = json.loads(entry_str)
                rr.append((service_type, client_id, service_id,
                    entry['mtime'], entry['ttl']))
            # sort by modification time
            # rr = sorted(rr, key=lambda entry: entry[3])
            r.extend(rr)
        return r
    # end get_all_clients

    # update publisher entry
    @cass_error_handler
    def update_service(self, service_type, service_id, entry):
        self.insert_service(service_type, service_id, entry)
    # end

    @cass_error_handler
    def insert_service(self, service_type, service_id, entry):<commit_msg>Set max columnns while reading cassandra db records.<commit_after>#
# Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
#

import re
from cfgm_common import jsonutils as json
import time
import gevent
import disc_consts
import disc_exceptions
from datetime import datetime
from gevent.coros import BoundedSemaphore

import pycassa
import pycassa.util
from pycassa.system_manager import *
from pycassa.util import *
from pycassa.types import *
from sandesh_common.vns.constants import DISCOVERY_SERVER_KEYSPACE_NAME, \
    CASSANDRA_DEFAULT_GC_GRACE_SECONDS

class DiscoveryCassandraClient(object):
    _DISCOVERY_KEYSPACE_NAME = DISCOVERY_SERVER_KEYSPACE_NAME
    _DISCOVERY_CF_NAME = 'discovery'

    @classmethod
    def get_db_info(cls):
        db_info = [(cls._DISCOVERY_KEYSPACE_NAME, [cls._DISCOVERY_CF_NAME])]
        return db_info
    # end get_db_info

    def __init__(self, module, cass_srv_list, reset_config=False,
                 max_retries=5, timeout=5, cass_credential=None):
        self._disco_cf_name = 'discovery'
        self._keyspace_name = 'DISCOVERY_SERVER'
        self._reset_config = reset_config
        self._credential = cass_credential
        self._cassandra_init(cass_srv_list, max_retries, timeout)

        self._debug = {
        }
    #end __init__

    # Helper routines for cassandra
    def _cassandra_init(self, server_list, max_retries, timeout):

        # column name <table-name>, <id1>, <id2>
        disco_cf_info = (self._disco_cf_name, 
            CompositeType(AsciiType(), UTF8Type(), UTF8Type()), AsciiType())

        # 1. Ensure keyspace and schema/CFs exist
        self._cassandra_ensure_keyspace(server_list, self._keyspace_name,
                [disco_cf_info])

        pool = pycassa.ConnectionPool(self._keyspace_name,
                                      server_list, max_overflow=-1,
                                      use_threadlocal=True, prefill=True,
                                      pool_size=100, pool_timeout=120,
                                      max_retries=max_retries, timeout=timeout,
                                      credentials=self._credential)
        rd_consistency = pycassa.cassandra.ttypes.ConsistencyLevel.ONE
        wr_consistency = pycassa.cassandra.ttypes.ConsistencyLevel.ONE
        self._disco_cf = pycassa.ColumnFamily(pool, self._disco_cf_name,
                                              read_consistency_level = rd_consistency,
                                              write_consistency_level = wr_consistency)
    #end _cassandra_init

    def _cassandra_ensure_keyspace(self, server_list,
                                   keyspace_name, cf_info_list):
        # Retry till cassandra is up
        server_idx = 0
        num_dbnodes = len(server_list)
        connected = False
        while not connected:
            try:
                cass_server = server_list[server_idx]
                sys_mgr = SystemManager(cass_server,credentials=self._credential)
                connected = True
            except Exception as e:
                # TODO do only for thrift.transport.TTransport.TTransportException
                server_idx = (server_idx + 1) % num_dbnodes
                time.sleep(3)

        if self._reset_config:
            try:
                sys_mgr.drop_keyspace(keyspace_name)
            except pycassa.cassandra.ttypes.InvalidRequestException as e:
                # TODO verify only EEXISTS
                print "Warning! " + str(e)

        try:
            # TODO replication_factor adjust?
            sys_mgr.create_keyspace(keyspace_name, SIMPLE_STRATEGY,
                                    {'replication_factor': str(num_dbnodes)})
        except pycassa.cassandra.ttypes.InvalidRequestException as e:
            # TODO verify only EEXISTS
            print "Warning! " + str(e)

        for cf_info in cf_info_list:
            try:
                (cf_name, comparator_type, validator_type) = cf_info
                sys_mgr.create_column_family(keyspace_name, cf_name, 
                        comparator_type = comparator_type, default_validation_class = validator_type)
                sys_mgr.alter_column_family(keyspace_name, cf_name,
                    gc_grace_seconds=CASSANDRA_DEFAULT_GC_GRACE_SECONDS)
            except pycassa.cassandra.ttypes.InvalidRequestException as e:
                # TODO verify only EEXISTS
                print "Warning! " + str(e)
    #end _cassandra_ensure_keyspace

    def get_debug_stats(self):
        return self._debug
    # end

    """
        various column names
        ('client', client_id, 'client-entry')
        ('subscriber', service_id, client_id)
        ('subscription', client_id, service_id)
        ('service', service_id, 'service-entry')
    """

    # decorator to catch connectivity error
    def cass_error_handler(func):
        def error_handler(*args, **kwargs):
            try:
                return func(*args,**kwargs)
            except (pycassa.pool.AllServersUnavailable,
                    pycassa.pool.MaximumRetryException):
                raise disc_exceptions.ServiceUnavailable()
            except Exception as e:
                raise
        return error_handler

    # return all publisher entries
    @cass_error_handler
    def service_entries(self, service_type = None):
        col_name = ('service',)
        try:
            data = self._disco_cf.get_range(column_start = col_name, column_finish = col_name)
            for service_type, services in data:
                for col_name in services:
                    col_value = services[col_name]
                    entry = json.loads(col_value)
                    col_name = ('subscriber', entry['service_id'],)
                    entry['in_use'] = self._disco_cf.get_count(service_type, 
                        column_start = col_name, column_finish = col_name)
                    yield(entry)
        except pycassa.pool.AllServersUnavailable:
            raise disc_exceptions.ServiceUnavailable()
            #raise StopIteration

    # return all clients
    def subscriber_entries(self):
        col_name = ('client',)
        data = self._disco_cf.get_range(column_start = col_name, column_finish = col_name)
        for service_type, clients in data:
            for col_name in clients:
                (foo, client_id, service_id) = col_name
                # skip pure client entry
                if service_id == disc_consts.CLIENT_TAG:
                    continue
                yield((client_id, service_type))
    # end

    # return all subscriptions
    @cass_error_handler
    def get_all_clients(self, service_type=None, service_id=None):
        r = []
        entry_format_subscriber = False
        if service_type and service_id:
            # ('subscriber', service_id, client_id)
            col_name = ('subscriber', service_id,)
            try:
                clients = self._disco_cf.get(service_type, column_start = col_name,
                    column_finish = col_name, column_count = disc_consts.MAX_COL)
            except pycassa.NotFoundException:
                return None
            data = [(service_type, dict(clients))]
            entry_format_subscriber = True
        elif service_type:
            col_name = ('client', )
            try:
                clients = self._disco_cf.get(service_type, column_start = col_name,
                    column_finish = col_name, column_count = disc_consts.MAX_COL)
            except pycassa.NotFoundException:
                return None
            data = [(service_type, dict(clients))]
        else:
            col_name = ('client', )
            try:
                data = self._disco_cf.get_range(column_start=col_name,
                   column_finish = col_name, column_count = disc_consts.MAX_COL)
            except pycassa.NotFoundException:
                return None

        for service_type, clients in data:
            rr = []
            for col_name in clients:
                if entry_format_subscriber:
                    (foo, service_id, client_id) = col_name
                else:
                    (foo, client_id, service_id) = col_name
                    # skip pure client entry
                    if service_id == disc_consts.CLIENT_TAG:
                        continue
                entry_str = clients[col_name]
                entry = json.loads(entry_str)
                rr.append((service_type, client_id, service_id,
                    entry['mtime'], entry['ttl']))
            # sort by modification time
            # rr = sorted(rr, key=lambda entry: entry[3])
            r.extend(rr)
        return r
    # end get_all_clients

    # update publisher entry
    @cass_error_handler
    def update_service(self, service_type, service_id, entry):
        self.insert_service(service_type, service_id, entry)
    # end

    @cass_error_handler
    def insert_service(self, service_type, service_id, entry):
{'fraction_non_alphanumeric': 0.07181208053691275, 'fraction_numerical': 0.0024608501118568234, 'mean_word_length': 1.9633742127941665, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7031412', 'n_tokens_mistral': 5205, 'n_tokens_neox': 4952, 'n_words': 1041}