<commit_before>#!/usr/bin/env python

#author <NAME>
#author <NAME>
#licence : apache v2


from pymongo import MongoClient
from pymongo import MongoReplicaSetClient
from pymongo.read_preferences import ReadPreference


import time

# for ceph command call
import subprocess

import datetime

import sys
import traceback
import os

import socket
from daemon import Daemon
 
import json
from StringIO import StringIO

from bson.dbref import DBRef 

from threading import Thread, Event

import httplib

import signal

# from bson.objectid import ObjectId
# db.col.find({"_id": ObjectId(obj_id_to_find)})

configfile = "/opt/inkscope/etc/inkscope.conf"
runfile = "/var/run/cephprobe/cephprobe.pid"
logfile = "/var/log/cephprobe.log"
clusterName = "ceph"
fsid = ""


# load the conf (from json into file)
def load_conf():
    datasource = open(configfile, "r")
    data = json.load(datasource)
    datasource.close()
    return data


# list sections prefixed
def ceph_conf_list(prefix):
    p = subprocess.Popen(
        args=[
            'ceph-conf',
            '-l',
            prefix
            ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    outdata, errdata = p.communicate()
    if (len(errdata)):
        raise RuntimeError('unable to get conf option prefix %s: %s' % (prefix, errdata))
    return outdata.rstrip().splitlines();


# get a field value from named section
def ceph_conf(field, name):
    p = subprocess.Popen(
        args=[
            'ceph-conf',
            '--show-config-value',
            field,
            '-n',
            name,
            ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    outdata, errdata = p.communicate()
    if len(errdata):
        raise RuntimeError('unable to get conf option %s for %s: %s' % (field, name, errdata))
    return outdata.rstrip()


# get a field value from global conf
def ceph_conf_global(field):
    p = subprocess.Popen(
        args=[
            'ceph-conf',
            '--show-config-value',
            field
            ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    outdata, errdata = p.communicate()
    if len(errdata):
        raise RuntimeError('unable to get conf option %s: %s' % (field, errdata))
    return outdata.rstrip()


# get a field value from global conf according to the specified ceph conf
def ceph_conf_global(cephConfPath, field):
    p = subprocess.Popen(
        args=[
            'ceph-conf',
            '-c',
            cephConfPath,
            '--show-config-value',
            field
            ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    outdata, errdata = p.communicate()
    if len(errdata):
        raise RuntimeError('unable to get conf option %s: %s' % (field, errdata))
    return outdata.rstrip()


# extract mons from conf and put them into mons
def process_conf():
    mon_sections=ceph_conf_list('mon.')
    if len(mon_sections)==0:
        initmon = ceph_conf_global('mon_initial_members')
        if not initmon:
            raise RuntimeError('enable to find a mon')
        mons = [initmon]
    else:
        for mon in mon_sections:
            mons.append(ceph_conf('host', mon))
        

# cluster
def init_cluster(restapi, ceph_rest_api_subfolder, db, hostname):
    leader = leadership(db, hostname)
    
    if isLeader or leader == None :  
        process_status(restapi, ceph_rest_api_subfolder, db)
        process_crushmap(restapi, ceph_rest_api_subfolder, db)
        process_osd_dump(restapi, ceph_rest_api_subfolder, db)
        process_pg_dump(restapi, ceph_rest_api_subfolder, db)
        process_df(restapi, ceph_rest_api_subfolder, db)
   
# health value
healthCst = ["HEALTH_OK", "HEALTH_WARN", "HEALTH_ERROR"]
healthMap = {}
for idx, h in enumerate(healthCst):
    healthMap[h] = idx
    
    
def worst_health(h1, h2):
    return healthCst[max(healthMap[h1], healthMap[h2])]

    
def leadership(db, hostname):    
    global isLeader
    leaderfailed = False
    cpleader = db.cephprobeleader.find_one()
    if cpleader :
        cp = db.cephprobe.find_one({"_id": cpleader["leader"]})
        if cp["timestamp"] < int(round((time.time()- (hb_refresh*2))*1000)) : 
            #leader failed !!
            leaderfailed = True
        else :
            isLeader = (cpleader["leader"] == hostname) #ensure leadership
            return cpleader["leader"]
    else :
        leaderfailed = True
        cpleader = {}
        
    if leaderfailed :
        isLeader = False
        # I am the new leader ?
        cephprobes = db.cephprobe.find( {"timestamp": {"$gt": int(round((time.time()- (hb_refresh*2)) * 1000))}})
        if cephprobes :
            cpids = [p["_id"] for p in cephprobes]
            cpids.sort()
            if cpids and (cpids[0] == hostname) : 
                # yes I am the new leader
                print "I'm the leader, then I work"
                sys.stdout.flush()
                cpleader["leader"] = hostname
                db.cephprobeleader.update({}, cpleader, upsert=True)   
                isLeader = True   
                return hostname
        else :
            # no one !!
            return None
                  
    

# uri : /api/v0.1/status.json
def process_status(restapi, ceph_rest_api_subfolder, db):
    if not isLeader :
        return
         
    print str(datetime.datetime.now()), "-- Process Status"  
    sys.stdout.flush()
    restapi.connect()
    restapi.request("GET", ceph_rest_api_subfolder+"/api/v0.1/status.json")
    r1=restapi.getresponse()
    if r1.status != 200:
        print str(datetime.datetime.now()), "-- error (Status) failed to connect to ceph rest api: ", r1.status, r1.reason
        restapi.close()
        return None
    else:
        data1 = r1.read()
        restapi.close()
        c_status = json.loads(data1)

        monmap = c_status['output']['monmap']
        
        
        map_stat_mon = {}
        health_services_list = c_status['output']['health']['health']['health_services']
       
        time_checks = c_status['output']['health']['timechecks']
        timecheckmap = {}
        try:
            for tc in time_checks["mons"]:
                tc["time_health"] = tc["health"]
                del tc["health"]
                monname = tc["name"]
                del tc["name"]
                timecheckmap[monname] = tc
        except (RuntimeError, TypeError, NameError, KeyError):
            pass

        # complete timecheck
        try:
            for health_service in health_services_list:
                health_services_mons = health_service['mons']
                for monst in health_services_mons:
                    monstat = monst.copy()
                    monstat["mon"] =  DBRef( "mon", monst['name'])
                    monstat["_id"] = monst['name']+":"+monst["last_updated"]
                    monstat["capacity_health"] = monstat["health"]

                    #complete with timecheck
                    if monstat["name"] in timecheckmap:
                        tc = timecheckmap[monstat["name"]]
                        monstat.update(tc)

                    monstat["health"] = worst_health(monstat["capacity_health"], monstat["time_health"])
                    del monstat["name"]
                    db.monstat.update({"_id" : monstat["_id"]}, monstat, upsert= True)
                    map_stat_mon[monst['name']] = monstat["_id"]
        except (RuntimeError, TypeError, NameError, KeyError):
            pass

        map_rk_name = {}
        
        for mon in monmap['mons']:
            
            #find the mon host
            hostaddr = mon['addr'].partition(':')[0]
            monhostid = None
            
            
            try :
                #monhostid = socket.gethostbyaddr(hostaddr)[0]
                monhostid = socket.getfqdn(hostaddr)
            except:                       
                monhost = db.hosts.find_one({"hostip": hostaddr})
    
                if hostaddr == '': # the case if mon is declared but not completly configured
                    monhostid = None
                elif not monhost:
                    monneti = db.net.find_one({"$where":  "this.inet.addr === '"+hostaddr+"'"})
                    if monneti:
                        monhostid = monneti["_id"].partition(":")[0]
                else:
                    monhostid = monhost["_id"]
            
            mondb = {"_id": mon['name'],
                     "host": DBRef( "hosts", monhostid), 
                     "addr": mon['addr'],
                     "rank": mon['rank'],
                    }
            
            if mon['name'] in map_stat_mon :
                mondb["stat"] = DBRef("monstat", map_stat_mon[mon['name']])
            db.mon.update({"_id": mon['name']}, mondb, upsert=True)
            map_rk_name[mon['rank']] = mon['name']        
            # no skew and latency ?
            
        mm = {"epoch": monmap['epoch'],
              "created": monmap['created'],
              "modified": monmap['modified'],
              "mons": [DBRef( "mon", m['name']) for m in monmap['mons']],
              "quorum": [DBRef( "mon", map_rk_name[rk]) for rk in c_status['output']['quorum']]
              }
        cluster = {"_id": c_status['output']['fsid'],
                   "election_epoch": c_status['output']['election_epoch'], 
                   "monmap": mm,
                   "pgmap": c_status['output']['pgmap'],
                   "osdmap-info": c_status['output']['osdmap']['osdmap'],
                   "name": clusterName, 
                   "health": c_status['output']['health']['overall_status'],
                   "health_detail": c_status['output']['health']['detail'],
                   "health_summary": c_status['output']['health']['summary']
                   }     
        db.cluster.update({'_id': c_status['output']['fsid']}, cluster, upsert=True)
        
        return c_status['output']['fsid']
   

# uri : /api/v0.1/osd/dump.json
def process_osd_dump(restapi, ceph_rest_api_subfolder, db):
    if not isLeader :
        return
    
    print str(datetime.datetime.now()), "-- Process OSDDump"  
    sys.stdout.flush()
    restapi.connect()
    restapi.request("GET", ceph_rest_api_subfolder+"/api/v0.1/osd/dump.json")
    r1=restapi.getresponse()
    if r1.status != 200:
        print str(datetime.datetime.now()), "-- error (OSDDump) failed to connect to ceph rest api: ", r1.status, r1.reason
        restapi.close()
    else:
        data1 = r1.read()
        restapi.close()
        osd_dump = json.loads(data1)

        osdsxinfo_map = {}
        for xi in osd_dump['output']['osd_xinfo']:
            osdsxinfo_map[xi["osd"]] = xi
        
        osds = osd_dump['output']['osds']
        
        for osd in osds:
            osd_stat = {"osd": DBRef("osd", osd["osd"]),
                        "timestamp": int(round(time.time() * 1000)),
                        "weight":  osd["weight"],
                        "up": osd["up"] == 1,
                        "in": osd["in"] == 1,
                        "last_clean_begin": osd["last_clean_begin"],
                        "last_clean_end": osd["last_clean_end"],
                        "up_from": osd["up_from"],
                        "up_thru": osd["up_thru"],
                        "down_at": osd["down_at"],
                        "lost_at": osd["lost_at"],
                        "state": osd["state"]
                        }
            osd_stat_id = db.osdstat.insert(osd_stat)
            
            
            hostaddr = osd["public_addr"].partition(':')[0]
            osdhostid = None
            
            #find host name
            try :
                #osdhostid = socket.gethostbyaddr(hostaddr)[0]
                osdhostid = socket.getfqdn(hostaddr)
            except:                       
                osdhost = db.hosts.find_one({"hostip": hostaddr})
    
                if hostaddr == '': # the case if osd is declared but not completly configured
                    osdhostid = None
                elif not osdhost:
                    osdneti = db.net.find_one({"$where":  "this.inet != null && this.inet.addr === '"+hostaddr+"'"})
                    if osdneti:
                        osdhostid = osdneti["_id"].partition(":")[0]
                else:
                    osdhostid = osdhost["_id"]
                    
                    
            
            osddatapartitionid = None
            if osdhostid:
                osddatapartition = db.partitions.find_one({"_id" : {'$regex' : osdhostid+":.*"}, "mountpoint" : '/var/lib/ceph/osd/'+clusterName+'-'+str(osd["osd"])})
                if osddatapartition :
                    osddatapartitionid = osddatapartition['_id']
                
            osddb = {"_id": osd["osd"],
                     "uuid": osd["uuid"],    
                     "node": DBRef( "nodes", osd["osd"]),
                     "stat":  DBRef( "osdstat", osd_stat_id),
                     "public_addr": osd["public_addr"],
                     "cluster_addr": osd["cluster_addr"],
                     "heartbeat_back_addr": osd["heartbeat_back_addr"],
                     "heartbeat_front_addr": osd["heartbeat_front_addr"],
                     "down_stamp": osdsxinfo_map[osd["osd"]]["down_stamp"],
                     "laggy_probability": osdsxinfo_map[osd["osd"]]["laggy_probability"],
                     "laggy_interval": osdsxinfo_map[osd["osd"]]["laggy_interval"],
                     "host":  DBRef( "hosts", osdhostid),
                     "partition": DBRef("partitions", osddatapartitionid)
                    }
            db.osd.update({'_id': osddb["_id"]}, osddb, upsert=True)
            
        pools = osd_dump['output']['pools']
        
        for pool in pools:
            p = pool.copy()
            p["_id"] = pool["pool"]
            del p["pool"]
            if p['auid'] : 
                p['auid'] = str(p['auid'])
            db.pools.update({'_id': p["_id"]}, p, upsert=True)


# osd host from conf : "host" : DBRef( "hosts", hostmap[i]),
# "partition" : DBRef( "partitions", hostmap[i]+":/dev/sdc1"),
# uri : /api/v0.1/pg/dump.json
def process_pg_dump(restapi, ceph_rest_api_subfolder, db):
    if not isLeader :
        return
    
    print str(datetime.datetime.now()), "-- Process PGDump"  
    sys.stdout.flush()
    restapi.connect()
    restapi.request("GET", ceph_rest_api_subfolder+"/api/v0.1/pg/dump.json")
    r1 = restapi.getresponse()
    if r1.status != 200:
        print str(datetime.datetime.now()), "-- error (PGDump) failed to connect to ceph rest api: ", r1.status, r1.reason
        restapi.close()
    else:
        data1 = r1.read()
        restapi.close()
        pgdump = json.loads(data1)
        for pg in pgdump["output"]["pg_stats"]:
            # db.pg.insert(pg)
            pg['_id'] = pg['pgid']
            del pg['pgid']
            pg['pool'] = DBRef('pools', int(pg['_id'].partition('.')[0]))
            
            ups = pg['up']
            pg['up'] = [DBRef('osd', i_osd) for i_osd in ups]
            
            actings = pg['acting']
            pg['acting'] = [DBRef('osd', i_osd) for i_osd in actings]
            
            # Rename keys containing '.' in stat_cat_sum 
            # replace '.' by '_'
            
            scs = pg['stat_cat_sum']
            for key in scs:
                try:
                    idx = key.index('.')
                    value = scs[key]
                    del scs[key]
                    scs[key.replace('.', '_')] = value
                except: 
                    pass
            
            db.pg.update({'_id' : pg["_id"]}, pg, upsert= True)<commit_msg>fix stat_cat_sum missing in 0.94 and later (#53)<commit_after>#!/usr/bin/env python

#author <NAME>
#author <NAME>
#licence : apache v2


from pymongo import MongoClient
from pymongo import MongoReplicaSetClient
from pymongo.read_preferences import ReadPreference


import time

# for ceph command call
import subprocess

import datetime

import sys
import traceback
import os

import socket
from daemon import Daemon
 
import json
from StringIO import StringIO

from bson.dbref import DBRef 

from threading import Thread, Event

import httplib

import signal

# from bson.objectid import ObjectId
# db.col.find({"_id": ObjectId(obj_id_to_find)})

configfile = "/opt/inkscope/etc/inkscope.conf"
runfile = "/var/run/cephprobe/cephprobe.pid"
logfile = "/var/log/cephprobe.log"
clusterName = "ceph"
fsid = ""


# load the conf (from json into file)
def load_conf():
    datasource = open(configfile, "r")
    data = json.load(datasource)
    datasource.close()
    return data


# list sections prefixed
def ceph_conf_list(prefix):
    p = subprocess.Popen(
        args=[
            'ceph-conf',
            '-l',
            prefix
            ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    outdata, errdata = p.communicate()
    if (len(errdata)):
        raise RuntimeError('unable to get conf option prefix %s: %s' % (prefix, errdata))
    return outdata.rstrip().splitlines();


# get a field value from named section
def ceph_conf(field, name):
    p = subprocess.Popen(
        args=[
            'ceph-conf',
            '--show-config-value',
            field,
            '-n',
            name,
            ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    outdata, errdata = p.communicate()
    if len(errdata):
        raise RuntimeError('unable to get conf option %s for %s: %s' % (field, name, errdata))
    return outdata.rstrip()


# get a field value from global conf
def ceph_conf_global(field):
    p = subprocess.Popen(
        args=[
            'ceph-conf',
            '--show-config-value',
            field
            ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    outdata, errdata = p.communicate()
    if len(errdata):
        raise RuntimeError('unable to get conf option %s: %s' % (field, errdata))
    return outdata.rstrip()


# get a field value from global conf according to the specified ceph conf
def ceph_conf_global(cephConfPath, field):
    p = subprocess.Popen(
        args=[
            'ceph-conf',
            '-c',
            cephConfPath,
            '--show-config-value',
            field
            ],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    outdata, errdata = p.communicate()
    if len(errdata):
        raise RuntimeError('unable to get conf option %s: %s' % (field, errdata))
    return outdata.rstrip()


# extract mons from conf and put them into mons
def process_conf():
    mon_sections=ceph_conf_list('mon.')
    if len(mon_sections)==0:
        initmon = ceph_conf_global('mon_initial_members')
        if not initmon:
            raise RuntimeError('enable to find a mon')
        mons = [initmon]
    else:
        for mon in mon_sections:
            mons.append(ceph_conf('host', mon))
        

# cluster
def init_cluster(restapi, ceph_rest_api_subfolder, db, hostname):
    leader = leadership(db, hostname)
    
    if isLeader or leader == None :  
        process_status(restapi, ceph_rest_api_subfolder, db)
        process_crushmap(restapi, ceph_rest_api_subfolder, db)
        process_osd_dump(restapi, ceph_rest_api_subfolder, db)
        process_pg_dump(restapi, ceph_rest_api_subfolder, db)
        process_df(restapi, ceph_rest_api_subfolder, db)
   
# health value
healthCst = ["HEALTH_OK", "HEALTH_WARN", "HEALTH_ERROR"]
healthMap = {}
for idx, h in enumerate(healthCst):
    healthMap[h] = idx
    
    
def worst_health(h1, h2):
    return healthCst[max(healthMap[h1], healthMap[h2])]

    
def leadership(db, hostname):    
    global isLeader
    leaderfailed = False
    cpleader = db.cephprobeleader.find_one()
    if cpleader :
        cp = db.cephprobe.find_one({"_id": cpleader["leader"]})
        if cp["timestamp"] < int(round((time.time()- (hb_refresh*2))*1000)) : 
            #leader failed !!
            leaderfailed = True
        else :
            isLeader = (cpleader["leader"] == hostname) #ensure leadership
            return cpleader["leader"]
    else :
        leaderfailed = True
        cpleader = {}
        
    if leaderfailed :
        isLeader = False
        # I am the new leader ?
        cephprobes = db.cephprobe.find( {"timestamp": {"$gt": int(round((time.time()- (hb_refresh*2)) * 1000))}})
        if cephprobes :
            cpids = [p["_id"] for p in cephprobes]
            cpids.sort()
            if cpids and (cpids[0] == hostname) : 
                # yes I am the new leader
                print "I'm the leader, then I work"
                sys.stdout.flush()
                cpleader["leader"] = hostname
                db.cephprobeleader.update({}, cpleader, upsert=True)   
                isLeader = True   
                return hostname
        else :
            # no one !!
            return None
                  
    

# uri : /api/v0.1/status.json
def process_status(restapi, ceph_rest_api_subfolder, db):
    if not isLeader :
        return
         
    print str(datetime.datetime.now()), "-- Process Status"  
    sys.stdout.flush()
    restapi.connect()
    restapi.request("GET", ceph_rest_api_subfolder+"/api/v0.1/status.json")
    r1=restapi.getresponse()
    if r1.status != 200:
        print str(datetime.datetime.now()), "-- error (Status) failed to connect to ceph rest api: ", r1.status, r1.reason
        restapi.close()
        return None
    else:
        data1 = r1.read()
        restapi.close()
        c_status = json.loads(data1)

        monmap = c_status['output']['monmap']
        
        
        map_stat_mon = {}
        health_services_list = c_status['output']['health']['health']['health_services']
       
        time_checks = c_status['output']['health']['timechecks']
        timecheckmap = {}
        try:
            for tc in time_checks["mons"]:
                tc["time_health"] = tc["health"]
                del tc["health"]
                monname = tc["name"]
                del tc["name"]
                timecheckmap[monname] = tc
        except (RuntimeError, TypeError, NameError, KeyError):
            pass

        # complete timecheck
        try:
            for health_service in health_services_list:
                health_services_mons = health_service['mons']
                for monst in health_services_mons:
                    monstat = monst.copy()
                    monstat["mon"] =  DBRef( "mon", monst['name'])
                    monstat["_id"] = monst['name']+":"+monst["last_updated"]
                    monstat["capacity_health"] = monstat["health"]

                    #complete with timecheck
                    if monstat["name"] in timecheckmap:
                        tc = timecheckmap[monstat["name"]]
                        monstat.update(tc)

                    monstat["health"] = worst_health(monstat["capacity_health"], monstat["time_health"])
                    del monstat["name"]
                    db.monstat.update({"_id" : monstat["_id"]}, monstat, upsert= True)
                    map_stat_mon[monst['name']] = monstat["_id"]
        except (RuntimeError, TypeError, NameError, KeyError):
            pass

        map_rk_name = {}
        
        for mon in monmap['mons']:
            
            #find the mon host
            hostaddr = mon['addr'].partition(':')[0]
            monhostid = None
            
            
            try :
                #monhostid = socket.gethostbyaddr(hostaddr)[0]
                monhostid = socket.getfqdn(hostaddr)
            except:                       
                monhost = db.hosts.find_one({"hostip": hostaddr})
    
                if hostaddr == '': # the case if mon is declared but not completly configured
                    monhostid = None
                elif not monhost:
                    monneti = db.net.find_one({"$where":  "this.inet.addr === '"+hostaddr+"'"})
                    if monneti:
                        monhostid = monneti["_id"].partition(":")[0]
                else:
                    monhostid = monhost["_id"]
            
            mondb = {"_id": mon['name'],
                     "host": DBRef( "hosts", monhostid), 
                     "addr": mon['addr'],
                     "rank": mon['rank'],
                    }
            
            if mon['name'] in map_stat_mon :
                mondb["stat"] = DBRef("monstat", map_stat_mon[mon['name']])
            db.mon.update({"_id": mon['name']}, mondb, upsert=True)
            map_rk_name[mon['rank']] = mon['name']        
            # no skew and latency ?
            
        mm = {"epoch": monmap['epoch'],
              "created": monmap['created'],
              "modified": monmap['modified'],
              "mons": [DBRef( "mon", m['name']) for m in monmap['mons']],
              "quorum": [DBRef( "mon", map_rk_name[rk]) for rk in c_status['output']['quorum']]
              }
        cluster = {"_id": c_status['output']['fsid'],
                   "election_epoch": c_status['output']['election_epoch'], 
                   "monmap": mm,
                   "pgmap": c_status['output']['pgmap'],
                   "osdmap-info": c_status['output']['osdmap']['osdmap'],
                   "name": clusterName, 
                   "health": c_status['output']['health']['overall_status'],
                   "health_detail": c_status['output']['health']['detail'],
                   "health_summary": c_status['output']['health']['summary']
                   }     
        db.cluster.update({'_id': c_status['output']['fsid']}, cluster, upsert=True)
        
        return c_status['output']['fsid']
   

# uri : /api/v0.1/osd/dump.json
def process_osd_dump(restapi, ceph_rest_api_subfolder, db):
    if not isLeader :
        return
    
    print str(datetime.datetime.now()), "-- Process OSDDump"  
    sys.stdout.flush()
    restapi.connect()
    restapi.request("GET", ceph_rest_api_subfolder+"/api/v0.1/osd/dump.json")
    r1=restapi.getresponse()
    if r1.status != 200:
        print str(datetime.datetime.now()), "-- error (OSDDump) failed to connect to ceph rest api: ", r1.status, r1.reason
        restapi.close()
    else:
        data1 = r1.read()
        restapi.close()
        osd_dump = json.loads(data1)

        osdsxinfo_map = {}
        for xi in osd_dump['output']['osd_xinfo']:
            osdsxinfo_map[xi["osd"]] = xi
        
        osds = osd_dump['output']['osds']
        
        for osd in osds:
            osd_stat = {"osd": DBRef("osd", osd["osd"]),
                        "timestamp": int(round(time.time() * 1000)),
                        "weight":  osd["weight"],
                        "up": osd["up"] == 1,
                        "in": osd["in"] == 1,
                        "last_clean_begin": osd["last_clean_begin"],
                        "last_clean_end": osd["last_clean_end"],
                        "up_from": osd["up_from"],
                        "up_thru": osd["up_thru"],
                        "down_at": osd["down_at"],
                        "lost_at": osd["lost_at"],
                        "state": osd["state"]
                        }
            osd_stat_id = db.osdstat.insert(osd_stat)
            
            
            hostaddr = osd["public_addr"].partition(':')[0]
            osdhostid = None
            
            #find host name
            try :
                #osdhostid = socket.gethostbyaddr(hostaddr)[0]
                osdhostid = socket.getfqdn(hostaddr)
            except:                       
                osdhost = db.hosts.find_one({"hostip": hostaddr})
    
                if hostaddr == '': # the case if osd is declared but not completly configured
                    osdhostid = None
                elif not osdhost:
                    osdneti = db.net.find_one({"$where":  "this.inet != null && this.inet.addr === '"+hostaddr+"'"})
                    if osdneti:
                        osdhostid = osdneti["_id"].partition(":")[0]
                else:
                    osdhostid = osdhost["_id"]
                    
                    
            
            osddatapartitionid = None
            if osdhostid:
                osddatapartition = db.partitions.find_one({"_id" : {'$regex' : osdhostid+":.*"}, "mountpoint" : '/var/lib/ceph/osd/'+clusterName+'-'+str(osd["osd"])})
                if osddatapartition :
                    osddatapartitionid = osddatapartition['_id']
                
            osddb = {"_id": osd["osd"],
                     "uuid": osd["uuid"],    
                     "node": DBRef( "nodes", osd["osd"]),
                     "stat":  DBRef( "osdstat", osd_stat_id),
                     "public_addr": osd["public_addr"],
                     "cluster_addr": osd["cluster_addr"],
                     "heartbeat_back_addr": osd["heartbeat_back_addr"],
                     "heartbeat_front_addr": osd["heartbeat_front_addr"],
                     "down_stamp": osdsxinfo_map[osd["osd"]]["down_stamp"],
                     "laggy_probability": osdsxinfo_map[osd["osd"]]["laggy_probability"],
                     "laggy_interval": osdsxinfo_map[osd["osd"]]["laggy_interval"],
                     "host":  DBRef( "hosts", osdhostid),
                     "partition": DBRef("partitions", osddatapartitionid)
                    }
            db.osd.update({'_id': osddb["_id"]}, osddb, upsert=True)
            
        pools = osd_dump['output']['pools']
        
        for pool in pools:
            p = pool.copy()
            p["_id"] = pool["pool"]
            del p["pool"]
            if p['auid'] : 
                p['auid'] = str(p['auid'])
            db.pools.update({'_id': p["_id"]}, p, upsert=True)


# osd host from conf : "host" : DBRef( "hosts", hostmap[i]),
# "partition" : DBRef( "partitions", hostmap[i]+":/dev/sdc1"),
# uri : /api/v0.1/pg/dump.json
def process_pg_dump(restapi, ceph_rest_api_subfolder, db):
    if not isLeader :
        return
    
    print str(datetime.datetime.now()), "-- Process PGDump"  
    sys.stdout.flush()
    restapi.connect()
    restapi.request("GET", ceph_rest_api_subfolder+"/api/v0.1/pg/dump.json")
    r1 = restapi.getresponse()
    if r1.status != 200:
        print str(datetime.datetime.now()), "-- error (PGDump) failed to connect to ceph rest api: ", r1.status, r1.reason
        restapi.close()
    else:
        data1 = r1.read()
        restapi.close()
        pgdump = json.loads(data1)
        for pg in pgdump["output"]["pg_stats"]:
            # db.pg.insert(pg)
            pg['_id'] = pg['pgid']
            del pg['pgid']
            pg['pool'] = DBRef('pools', int(pg['_id'].partition('.')[0]))
            
            ups = pg['up']
            pg['up'] = [DBRef('osd', i_osd) for i_osd in ups]
            
            actings = pg['acting']
            pg['acting'] = [DBRef('osd', i_osd) for i_osd in actings]
            
            # Rename keys containing '.' in stat_cat_sum 
            # replace '.' by '_'
            if 'stat_cat_sum' in pg:
                scs = pg['stat_cat_sum']
            else:
                scs = pg['stat_sum']

            for key in scs:
                try:
                    idx = key.index('.')
                    value = scs[key]
                    del scs[key]
                    scs[key.replace('.', '_')] = value
                except: 
                    pass
            
            db.pg.update({'_id' : pg["_id"]}, pg, upsert= True)
{'fraction_non_alphanumeric': 0.13241575700047462, 'fraction_numerical': 0.0047777250435057745, 'mean_word_length': 1.7485868336377077, 'pattern_counts': {'":': 114, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1522134', 'n_tokens_mistral': 9913, 'n_tokens_neox': 9527, 'n_words': 1983}