ioids.dataengine

26

27 - def __init__(self):

28 """ 29 Sets up the queues for incoming messages. 30 """ 31 self._localEvents = [] 32 self._localIoidsEvents =[] 33 self._remoteEvents = [] 34 self._running = 0

35

36 - def startup(self):

37 """ 38 Start up the data engine in its background thread. 39 """ 40 from config import DATA_ENGINE_PROCESSING_INTERVAL 41 self._interval = DATA_ENGINE_PROCESSING_INTERVAL 42 self._running = 1 43 import thread 44 thread.start_new_thread(self.runUntilShutdown, ()) 45 ## self.runUntilShutdown() 46 47 from ioidslogging import DATAENGINE_STATUS, getDefaultLogger 48 getDefaultLogger().newMessage(DATAENGINE_STATUS, 'Data engine process started')

49 50

51 - def runUntilShutdown(self):

52 """ 53 In here, all the queues are checked frequently and appropriate actions are undertaken. 54 """ 55 from ioidslogging import DATAENGINE_PROCESSING_DETAILS, getDefaultLogger, DATAENGINE_ERROR_GENERIC 56 57 import time 58 time.sleep(self._interval) 59 while self._running: 60 try: 61 getDefaultLogger().newMessage(DATAENGINE_PROCESSING_DETAILS, 'Data engine details: check my lists') 62 counter = 0 63 while len(self._localEvents): 64 self._processEventFromLocal(self._localEvents[0]) 65 del self._localEvents[0] 66 counter += 1 67 getDefaultLogger().newMessage(DATAENGINE_PROCESSING_DETAILS, '-- Data engine details: Processed %d local events.' %(counter)) 68 counter = 0 69 while len(self._localIoidsEvents): 70 self._processIoidsEventFromLocal(self._localIoidsEvents[0]) 71 del self._localIoidsEvents[0] 72 counter += 1 73 getDefaultLogger().newMessage(DATAENGINE_PROCESSING_DETAILS, '-- Data engine details: Processed %d local ioids events.' %(counter)) 74 ## except Exception, msg: 75 except ValueError, msg: 76 getDefaultLogger().newMessage(DATAENGINE_ERROR_GENERIC, 'Data engine ERROR: %s' %(msg)) 77 time.sleep(self._interval)

78

79 - def shutdown(self):

80 """ 81 Shutdown the data engine thread. 82 """ 83 self._running = 0 84 from ioidslogging import DATAENGINE_STATUS, getDefaultLogger 85 getDefaultLogger().newMessage(DATAENGINE_STATUS, 'Data engine process stopped')

86

87 - def _executeOneReaction(self, event, reaction):

88 """ 89 Performs all operations as defined by the reaction part of an ioids rule. 90 """ 91 from config import G4DS_MEMBER_ID 92 from dbconnector import getDBConnector 93 from errorhandling import IoidsDependencyException 94 95 ioidsSource = G4DS_MEMBER_ID 96 ioidsSender = G4DS_MEMBER_ID 97 if reaction['parameters'].has_key('community'): 98 if reaction['parameters']['community'] == 'Auto': 99 ioidsCommunity = 'C001' # we will do this properly soon :) TODO 100 else: 101 ioidsCommunity = reaction['parameters']['community'] 102 else: 103 raise IoidsDependencyException('Community can not be determined for new local event. Looks like a mistake in ioids policy.') 104 105 if reaction['parameters'].has_key('classification'): 106 if reaction['parameters']['classification'] == 'Auto': 107 ioidsClassificationCode = '10' # we will do this properly soon :) TODO 108 else: 109 ioidsClassificationCode = reaction['parameters']['classification'] 110 else: 111 raise IoidsDependencyException('Community can not be determined for new local event. Looks like a mistake in ioids policy.') 112 113 ioidsTimestamp = 'now' 114 115 116 if reaction['type'] == 'NewLocalEvent': 117 if event[1].has_key('event_id'): # we must get rid off the id - otherwise it will insert a new event again and again 118 del event[1]['event_id'] 119 120 # create relations 121 from dataengine_tools import getPreXMLDictCreator 122 from config import IOIDS_EVENT_TYPE, LOCAL_ADDRESS, LOCAL_HOSTNAME, LOCAL_MAC, LOCAL_OS, LOCAL_DOMAIN, LOCAL_COMPUTER_TYPE 123 from messagewrapper import getXMLDBWrapper 124 import binascii as hex 125 creator = getPreXMLDictCreator() 126 127 # here we create the actual event 128 newEncoding = creator.createNewEncodingEntry('XML HEX') 129 eventXML = getXMLDBWrapper().wrapInsert(event[0], event[1], event[2]) 130 encoded = hex.hexlify(eventXML) 131 newData = creator.createNewDataEntry(encoded, [newEncoding]) # todo: put whole event description here 132 133 newComputer = creator.createNewComputerEntry(LOCAL_HOSTNAME, LOCAL_OS, LOCAL_ADDRESS, LOCAL_MAC, LOCAL_DOMAIN, [], None, LOCAL_COMPUTER_TYPE) 134 newAgent = creator.createNewAgentEntry('IOIDS', [newComputer], '2') 135 newReporter = creator.createNewReporterEntry('IOIDS reporter', [newAgent]) 136 137 newEventType = creator.createNewEventTypeEntry(IOIDS_EVENT_TYPE) 138 139 # reporter is me 140 # observer is the reporter from our event 141 oldEventReporterId = event[1]['rprt_id'] 142 fullReporter = getDBConnector().getReporter(oldEventReporterId) 143 if fullReporter[1].has_key('rprt_name'): 144 repName = fullReporter[1]['rprt_name'] 145 else: 146 repName = None 147 newObserver = creator.createNewObserverEntry(repName, fullReporter[2]) 148 # source and destination are the same than of the actual event 149 newEvent = creator.createNewEventEntry('now', [newData, newEventType, newReporter, newObserver], None, None, 150 event[1]['src_id'], event[1]['dstn_id']) 151 ioidsEventEntry = creator.createNewIoidsEventEntry(ioidsCommunity, ioidsTimestamp, [ 152 creator.createNewIoidsSourceEntry(ioidsSource), 153 creator.createNewIoidsSenderEntry(ioidsSender), 154 getDBConnector().getIoidsClassificationByCode(ioidsClassificationCode), 155 ## creator.createNewIoidsClassificationEntry(ioidsClassificationCode, ioidsClassificationName), 156 newEvent # our event should be in the proper format already 157 ]) 158 ## creator.createIoidsClassificationEntry(ioidsClassification)], event['event_id']) 159 160 # and finally the relations 161 newRelationEntry = creator.createNewIoidsRelationEntry([ioidsEventEntry, event], relationTypeName = 'parent') 162 163 # testing purposes 164 ## import support.dictviewer 165 ## support.dictviewer.showNow(newRelationEntry) 166 # #### 167 168 primKeyRel = getDBConnector().insertFullIoidsEventWithRelation(newRelationEntry) 169 ## ioidsEventId = getDBConnector().getIoidsRelation(primKeyRel,0)[1]['ioids_event_id'] 170 ## ## primKey = getDBConnector().insertIoidsEvent(ioidsEventEntry) 171 ## eventId = getDBConnector().getIoidsEvent(ioidsEventId, 0)[1]['event_id'] 172 ## self._remoteEvents.append(eventId) 173 print "\t-- Inserted event with id: %s" %(primKeyRel) 174 175 # now let's go and check whether this is to be distributed 176 if reaction['parameters'].has_key('distribute'): 177 print "\t--Now I would even send it off to %s." %(reaction['parameters']['distribute']['domain'])

178 # but that's for tomorrow ;) 179

180 - def _processEventFromLocal(self, event):

181 """ 182 Processes one item from the local event list. 183 """ 184 from config import G4DS_MEMBER_ID 185 from dbconnector import getDBConnector 186 187 ## 188 ## real dataengine started here 189 ## 190 191 event1 = getDBConnector().getEvent(event[1]['event_id']) 192 subsystem = None 193 for rel in event1[2]: 194 if rel[0] == 'event_type': 195 subsystem = rel[1]['event_type_name'] 196 print "Local event - subsystem determination: %s" %(subsystem) 197 198 # that's all we need here - let's ask for the corrospondig policies 199 from policyengine import getPolicyEngine 200 params = {} 201 params['origin'] = 'local' 202 if subsystem: 203 params['subsystem'] = subsystem 204 reactions = getPolicyEngine().lookup(params) 205 206 # let's carry out the reactions then 207 for reaction in reactions: 208 print "\t-- Carry out reaction # %s: %s" %(reaction['id'], reaction['type']) 209 self._executeOneReaction(event, reaction) 210 211 return

212 213 214 215 216

217 - def _processIoidsEventFromLocal(self, event):

218 from dbconnector import getDBConnector 219 ioidsevent = getDBConnector().getIoidsEvent(event[1]['ioids_event_id']) 220 ## print ioidsevent 221 from messagewrapper import getXMLDBWrapper, getIoidsMessageWrapper 222 x = getXMLDBWrapper().wrapInsert(ioidsevent[0], ioidsevent[1], ioidsevent[2]) 223 ## print "***** SENT:\n", x 224 from g4dsconnector import getG4dsConnector 225 ## getG4dsConnector().sendMessage(ioidsevent) 226 227 relatedEvents = getDBConnector().getRelatedEventsForIoidsEvent(event[1]['ioids_event_id']) 228 # determine the extension information for each related event 229 for relEvent in relatedEvents: 230 relEventEvent = getIoidsMessageWrapper()._getRelationInTree(relEvent, ['event']) 231 extName, extValue = getDBConnector().getExtensionForEvent(relEventEvent) 232 dict = {} 233 dict['extension_name'] = extName 234 relEvent[2].append(['extension',dict, [extValue]]) 235 236 from messagewrapper import getIoidsMessageWrapper 237 xml = getIoidsMessageWrapper().assembleIoidsMessage(ioidsevent, relatedEvents) 238 ## print "Sending:\n%s" %xml 239 240 getG4dsConnector().sendEventUpdate(xml) 241 print "*** processed (and sent) IOIDS event %s" %(event[1]['ioids_event_id'])

242 243 # 244 # 245 # testing 246 # 247 ## print "*" * 40, 'TEST', "*" * 40 248 ## xmlNew = getIoidsMessageWrapper().wrapKnowledgeReplyMessage([(ioidsevent, relatedEvents)]) 249 ## print xmlNew 250 ## print "*" * 40, 'TEST', "*" * 40 251 ## newInDict = getIoidsMessageWrapper().parseKnowledgeReplyMessage(xmlNew) 252 ## print "%s\n%s" %(newInDict[0][0], newInDict[0][1]) 253 ## print "*" * 40, 'TEST', "*" * 40 254

255 - def newEventFromLocal(self, event):

256 """ 257 Processes the occurence of a new event in the local database. 258 259 Should be called from the ioids event trigger. 260 """ 261 print "Received event (local) with id: %s - put it into event queue." %(event[1]['event_id']) 262 self._localEvents.append(event)

263

264 - def newIoidsEventFromLocal(self, ioidsevent):

265 """ 266 Processes the occurence of a new ioids event in the local database. 267 268 Should be called from the ioids event trigger. 269 """ 270 print "Received ioids event (local) with id: %s - put into ioids event queue." %(ioidsevent[1]['ioids_event_id']) 271 self._localIoidsEvents.append(ioidsevent)

272

273 - def newIoidsEventFromRemote(self, ioidsevent, relations = []):

274 from dbconnector import getDBConnector 275 ## print "I received from remote:\nEvent: %s\nRelations: %s" %(event, relations) 276 print "I received from remote Event with Relations" 277 primKey = getDBConnector().insertIoidsEvent(ioidsevent) 278 eventId = getDBConnector().getIoidsEvent(primKey, 0)[1]['event_id'] 279 self._remoteEvents.append(eventId) # our trigger must not pick up this event 280 281 from dataengine_tools import getPreXMLDictCreator 282 for relation in relations: 283 print "New Relation:" 284 plainEvent = None 285 extensionEvent = None 286 relationType = relation[1]['type'] 287 extensionType = None 288 for entry in relation[2]: 289 if entry[0] == 'plainevent': 290 plainEvent = entry[2][0] 291 elif entry[0] == 'extension': 292 try: 293 extensionEvent = entry[2][0] 294 extensionType = entry[1]['type'] 295 except IndexError, msg: 296 pass # no prob, that only means, that the sender could not handle the extension 297 298 299 relType = getPreXMLDictCreator().createNewIoidsRelationTypeEntry(relationType) 300 relEntry = getPreXMLDictCreator().createNewIoidsRelationEntry([ioidsevent, plainEvent, relType]) 301 302 # testing purposes 303 ## import support.dictviewer 304 ## support.dictviewer.showNowAscii(relEntry) 305 #### support.dictviewer.showNow(relEntry) 306 # #### 307 primKey = getDBConnector().insertFullIoidsEventWithRelation(relEntry) 308 print "-- Primary key for remote ioids event (relation): %s" %(primKey) 309 310 print "-- Event for Extension: %s" %(extensionType) 311 ## support.dictviewer.showNowAscii(extensionEvent) 312 try: 313 primKey = getDBConnector().insertExtensionEvent(extensionType, extensionEvent) 314 print "-- Primary key for extension event: %s" %(primKey) 315 except ValueError, msg: 316 print "-- Extension is unknown: %s" %(extensionType) 317 pass # that's fine again - only means that I myself do not understand the extension here.

Source Code for Module ioids.dataengine