ltsp-cluster-lbserver 2.0.0-0ubuntu5 / usr / lib / python2.7 / dist-packages / lbserver / lbserver.py

#!/usr/bin/env python

"""
LBServer is a service to be executed on the load balancer server.
"""

import sys
import random
import xmlrpclib
from threading import Thread

debug=False

class LBServer:
    def __init__(self):
        self.groups={}
        self.variables=[]
        self.lblisten=('',8008)
        self.mgmtlisten=('',8001)
        self.mgmtenabled=False
        self.isslave=False
        self.masterurl=None
        self.maxthreads=10
        self.refresh=60
        self.cfgrefresh=60
        self.timeout=10
        self.returns="IP"
        self.getters=None
    def _activeGetters(self):
        nbActive = 0
        for getter in self.getters:
            if getter.isAlive():
                nbActive += 1
    def getStatusStruct(self):
        return [g.getStruct() for g in self.groups.values()]
    def getNodes(self):
        nodes = []
        for g in self.groups.values():
            nodes += g.nodes
        return nodes
    def refreshNodesValues(self):
        self.getters=[]
        for node in self.getNodes():
            while self._activeGetters() >= self.maxthreads:
                time.sleep(1)
            getter = LBNodeGetter(node)
            self.getters.append(getter)
            getter.start()
        for getter in self.getters:
            getter.join(self.timeout)
        self.getters=None
    def getNextServer(self):
        for g in self.groups.values():
            if g.default:
                return self.getNextServerForGroup(g.name)
    def getNextServerForGroup(self,grpName):
        # If group not found, use default
        if not(self.groups.has_key(grpName)):
            return self.getNextServer()
        # Use flags to ensure groups visited only one time
        flags={}
        grp=self.groups[grpName]
        while grp != None and not(flags.has_key(grp.name)):
            nextServer=grp.getNextServer()
            if nextServer == None:
                flags[grp.name]=True
                if grp.backup!=None and self.groups.has_key(grp.backup):
                    grp=self.groups[grp.backup]
            else:
                return nextServer
        return None


class LBGroup:
    """
    LBGroup
    Contains all node information
    """
    def __init__(self,lbServer):
        self._lbServer=lbServer
        self.nodes=[]
        self.name=None
        self.default=False
        self.backup=None
    def getStruct(self):
        self.getNextServer() # Compute weights
        strt = {}
        strt["name"]=self.name
        strt["default"]=self.default
        if self.backup: strt["backup"]=self.backup
        strt["nodes"] = [n.getStruct() for n in self.nodes]
        return strt
    def getNextServer(self):
        # Create active nodes list
        tabNodes=[]
        for n in self.nodes:
            if n.active:
                tabNodes.append(n)
        if len(tabNodes)==0:
            return None
        if debug:
            print "==getNextServer on nodes :"
            for x in tabNodes:
                print x.name
            print "=="
        # Identify values to use and summarize free values
        mapVar={}
        mapSumFree={}
        for var in self._lbServer.variables:
            tabVal=[]
            sumFree=0
            for n in tabNodes:
                if n.values.has_key(var.name):
                    val = n.values[var.name]
                    tabVal.append(val)
                    if val.free != None:
                        sumFree += val.free
            if len(tabVal)>0 and var.weight>0:
                mapVar[var.name]=tabVal
                mapSumFree[var.name]=sumFree
        if debug:
            print "==Variables list :"
            for x in mapVar.keys():
                print "Var : %s sumfree : %s weight : %s"%(x,mapSumFree[x],mapVar[x][0].weight)
            print "=="
        # Compute weight of all nodes
        tabNodesWeight=[]
        for i in range(len(tabNodes)):
            tabNodesWeight.append(0)
            for varName in mapVar.keys():
                if tabNodes[i].values.has_key(varName):
                    val = tabNodes[i].values[varName]
                    if val.free > 0 and mapSumFree[varName] > 0:
                        valWeight = int(round((float(val.free)/float(mapSumFree[varName]))*val.weight))
                        tabNodesWeight[i] += valWeight
                        if debug:
                            print "Node : %s var : %s val : %s capacity : %s critical : %s valfree : %s weight : %s"%(tabNodes[i].name,varName,val.value,val.capacity,val.critical,val.free,valWeight)
            tabNodes[i].lastWeight = tabNodesWeight[i]
            if debug:
                print "Node : %s total weight : %s"%(tabNodes[i].name,tabNodesWeight[i])
        # Compute total Weight
        sumWeight=0
        for w in tabNodesWeight: sumWeight+=w
        if sumWeight==0:
            return None
        # Choose a random node, probability depend on node Weight
        rnd=random.randint(1,sumWeight)
        idx=0
        nodeId=-1
        while idx < rnd:
            nodeId+=1
            idx+=tabNodesWeight[nodeId]
        return tabNodes[nodeId].values[self._lbServer.returns].value


class LBNode:
    def __init__(self,lbServer):
        self._lbServer=lbServer
        self.name=None
        self.address=None
        self.active=False
        self.values = {}
        self.lastWeight=None
    def parseString(self,map,s):
        s = str(s)
        # Sort by name length
        sTab = [] + map.values()
        sTab.sort(lambda v1, v2: cmp(len(v2.name),len(v1.name)))
        for x in sTab:
            if not(isinstance(x.value,list)):
                s = s.replace('$'+x.name,str(x.value))
        return s
    def evalExpr(self,map,s,exprName='expression'):
        try:
            val = eval(self.parseString(map,s))
        except:
            raise Exception, "%s: '%s', eval: '%s'"%(exprName,s,self.parseString(map,s))
        return val
    def getStruct(self):
        strt = {}
        strt["name"] = self.name
        strt["address"] = self.address
        strt["active"] = self.active
        if self.lastWeight: strt["lastWeight"] = self.lastWeight
        strt["values"] = [v.getStruct() for v in self.values.values()]
        return strt
    def updateValues(self,map):
        # Update variables, set weight to used variables
        active = True
        for var in self._lbServer.variables:
            if map.has_key(var.name):
                nodeVal = map[var.name]
                if isinstance(nodeVal.value,str):
                    nodeVal.value = eval(nodeVal.value)
                for rule in var.rules:
                    errorDetail=""
                    try:
                        if rule.test==None or self.evalExpr(map,rule.test,"rule")==True:
                            if nodeVal.capacity==None and rule.capacity!=None:
                                nodeVal.capacity=self.evalExpr(map,rule.capacity,"capacity")
                            if nodeVal.critical==None and rule.critical!=None:
                                nodeVal.critical=self.evalExpr(map,rule.critical,"critical")
                            if nodeVal.capacity!=None:
                                nodeVal.free=nodeVal.capacity-nodeVal.value
                                if nodeVal.free < 0: 
                                    nodeVal.free = 0
                                nodeVal.weight=var.weight
                            # Disable node with a rule in critical state
                            if nodeVal.value >= nodeVal.critical:
                                print "Node '%s' disabled, %s: %s >= %s"%(self.name,var.name,nodeVal.value,nodeVal.critical)
                                active = False
                            break
                    except Exception, e:
                        print "Error updating node : %s, %s"%(self.name,sys.exc_info()[1])
        self.active=active
        self.values=map


class LBNodeGetter(Thread):
    def __init__(self,node):
        Thread.__init__(self)
        self._node=node
    def run(self):
        # Retrieve node values
        tabValues=[]
        try:
            rpcserver = xmlrpclib.ServerProxy(self._node.address)
            tabValues=rpcserver.get_values()
        except:
            print "Cannot retrieve node '%s', error:%s"%(self._node.name,sys.exc_info()[0])
            self._node.active=False
            return
        # Update node values
        try:
            values = {}
            for v in tabValues:
                nodeVal=LBNodeValue(self._node,v)
                values[nodeVal.name]=nodeVal
            self._node.updateValues(values)
        except Exception:
            self._node.active=False
            raise


class LBNodeValue:
    """
    NodeValue
    Contains status information
    """
    def __init__(self,node,struct):
        self._node=node
        self.name=struct["name"]
        self.value=struct["value"]
        self.capacity=None
        self.critical=None
        if struct.has_key("capacity"):
            self.capacity=struct["capacity"]
        if struct.has_key("critical"):
            self.critical=struct["critical"]
        self.free=None
        self.weight=None
    def getStruct(self):
        strt = {}
        strt["name"]=self.name
        if self.value:
            strt["value"]=self.value
            strt["critic"]=(self.critical != None and self.value >= self.critical)
        return strt


class LBVariable:
    """
    Variable
    """
    def __init__(self):
        self.name=None
        self.weight=50
        self.rules=[]


class LBRule:
    """
    LBRule
    """
    def __init__(self):
        self.test=None
        self.capacity = None
        self.critical = sys.maxint


def main():
    pass

if __name__ == "__main__":
    main()