Skip to content

Commit

Permalink
Improved objects and events custom handler dispatching
Browse files Browse the repository at this point in the history 
Signed-off-by: Christian Pinto <christian.pinto@ibm.com>
  • Loading branch information
@christian-pinto
christian-pinto committed Feb 28, 2024
1 parent 963736e commit aee63d8
Show file tree
Hide file tree
Showing 2 changed files with 136 additions and 86 deletions.
176 changes: 101 additions & 75 deletions sunfish/events/redfish_event_handler.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,96 @@
from sunfish.events.redfish_subscription_handler import subscribtions
from sunfish.lib.exceptions import *

logger = logging.getLogger("RedfishEventHandler")


class RedfishEventHandlersTable:
@classmethod
def AggregationSourceDiscovered(cls, event_handler: EventHandlerInterface, event: dict, context: str):
###
# Fabric Agents are modelled as AggregationSource objects (RedFish v2023.1 at the time of writing this comment)
# Registration will happen with the OFMF receiving a and event with MessageId: AggregationSourceDiscovered
# The arguments of the event message are:
# - Arg1: "Redfish"
# - Arg2: "agent_ip:port"
# I am also assuming that the agent name to be used is contained in the OriginOfCondifiton field of the event as in the below example:
# {
# "OriginOfCondition: [
# "@odata.id" : "/redfish/v1/AggregationService/AggregationSource/AgentName"
# ]"
# }
logger.info("AggregationSourceDiscovered method called")

connectionMethodId = event['OriginOfCondition']['@odata.id']
hostname = event['MessageArgs'][1] # Agent address

response = requests.get(f"{hostname}/{connectionMethodId}")
if response.status_code != 200:
raise Exception("Cannot find ConnectionMethod")
response = response.json()

### Save agent registration
connection_method_name = connectionMethodId.split('/')[-1]
connection_method_name = connectionMethodId[:-len(connection_method_name)]
event_handler.core.create_object(connection_method_name, response)

connection_method_template = {
"@odata.type": "#AggregationSource.v1_2_.AggregationSource",
"HostName": hostname,
"Links": {
"ConnectionMethod": {
"@odata.id": connectionMethodId
},
"ResourcesAccessed": []
}
}

try:
resp_post = event_handler.core.create_object(
os.path.join(event_handler.core.conf["redfish_root"], "AggregationService/AggregationSources"),
connection_method_template)
except Exception:
raise Exception()

aggregation_source_id = resp_post['@odata.id']
agent_subscription_context = {"Context": aggregation_source_id.split('/')[-1]}

resp_patch = requests.patch(f"{hostname}/redfish/v1/EventService/Subscriptions/SunfishServer",
json=agent_subscription_context)

return resp_patch

@classmethod
def ResourceCreated(cls, event_handler: EventHandlerInterface, event: dict, context: str):
logger.info("New resource created")

id = event['OriginOfCondition']['@odata.id'] # /redfish/v1/Fabrics/CXL
aggregation_source = event_handler.core.storage_backend.read(
os.path.join(event_handler.core.conf["redfish_root"],
"AggregationService", "AggregationSources", context)
)
hostname = aggregation_source["HostName"]

response = requests.get(f"{hostname}/{id}")
if response.status_code != 200:
raise Exception("Cannot find ConnectionMethod")
response = response.json()

add_aggregation_source_reference(response, aggregation_source)

event_handler.core.create_object(id, response)

RedfishEventHandler.bfsInspection(event_handler.core, response, aggregation_source)

event_handler.core.storage_backend.patch(id, aggregation_source)


class RedfishEventHandler(EventHandlerInterface):
dispatch_table = {
"AggregationSourceDiscovered": RedfishEventHandlersTable.AggregationSourceDiscovered,
"ResourceCreated": RedfishEventHandlersTable.ResourceCreated
}

def __init__(self, core):
"""init that sets the conf and calls the load subcriptions method
Expand All @@ -23,6 +112,12 @@ def __init__(self, core):
self.fs_root = core.conf["backend_conf"]["fs_root"]
self.subscribers_root = core.conf["backend_conf"]["subscribers_root"]
self.backend = core.storage_backend
@classmethod
def dispatch(cls, message_id: str, event_handler: EventHandlerInterface, event: dict):
if message_id in cls.dispatch_table:
return cls.dispatch_table[message_id](event_handler, event)
else:
logger.debug(f"Message id '{message_id}' does not have a custom handler")

def new_event(self, payload):
"""Compares event's information with the subsribtions data structure to find the Ids of the subscribers for that event.
Expand Down Expand Up @@ -138,76 +233,6 @@ def check_subdirs(self, origin):
to_forward.append(id)

return to_forward

def AggregationSourceDiscovered(self, event, context):
###
# Fabric Agents are modelled as AggregationSource objects (RedFish v2023.1 at the time of writing this comment)
# Registration will happen with the OFMF receiving a and event with MessageId: AggregationSourceDiscovered
# The arguments of the event message are:
# - Arg1: "Redfish"
# - Arg2: "agent_ip:port"
# I am also assuming that the agent name to be used is contained in the OriginOfCondifiton field of the event as in the below example:
# {
# "OriginOfCondition: [
# "@odata.id" : "/redfish/v1/AggregationService/AggregationSource/AgentName"
# ]"
# }
logging.info("AggregationSourceDiscovered method called")

connectionMethodId = event['OriginOfCondition']['@odata.id']
hostname = event['MessageArgs'][1] # Agent address

response = requests.get(f"{hostname}/{connectionMethodId}")
if response.status_code != 200:
raise Exception("Cannot find ConnectionMethod")
response = response.json()

### Save agent registration
connection_method_name = connectionMethodId.split('/')[-1]
connection_method_name = connectionMethodId[:-len(connection_method_name)]
self.create_object(connection_method_name, response)

connection_method_template = {
"@Redfish.Copyright": "Copyright 2014-2021 SNIA. All rights reserved.",
"@odata.type": "#AggregationSource.v1_2_.AggregationSource",
"HostName": hostname,
"Links": {
"ConnectionMethod": {
"@odata.id": connectionMethodId
},
"ResourcesAccessed": [ ]
}
}

resp_post = self.create_object(os.path.join(self.conf["redfish_root"], "AggregationService/AggregationSources"), connection_method_template)

aggregation_source_id = resp_post['@odata.id']
agent_subscription_context = {"Context": aggregation_source_id.split('/')[-1]}

resp_patch = requests.patch(f"{hostname}/redfish/v1/EventService/Subscriptions/SunfishServer",
json=agent_subscription_context)

return resp_patch

def ResourceCreated(self, event, context):
logging.info("New resource created")

id = event['OriginOfCondition']['@odata.id'] # /redfish/v1/Fabrics/CXL
aggregation_source = self.get_object(
os.path.join(self.conf["redfish_root"], "AggregationService", "AggregationSources", context))
hostname = aggregation_source["HostName"]

response = requests.get(f"{hostname}/{id}")
if response.status_code != 200:
raise Exception("Cannot find ConnectionMethod")
object = response.json()
add_aggregation_source_reference(object, aggregation_source)

self.create_object(id, object)

RedfishEventHandler.bfsInspection(self, object, aggregation_source)

self.patch_object(id, aggregation_source)

def bfsInspection(self, node, aggregation_source):
queue = []
Expand All @@ -234,7 +259,7 @@ def handleNestedObject(self, obj):
redfish_obj = RedfishEventHandler.fetchResourceAndTree(self, id, aggregation_source, visited, queue, fetched)

if redfish_obj is None or type(redfish_obj) != dict:
logging.info(f"Resource - {id} - not available")
logger.info(f"Resource - {id} - not available")
continue

for key, val in redfish_obj.items():
Expand Down Expand Up @@ -274,10 +299,10 @@ def fetchResourceAndTree(self, id, aggregation_source, visited, queue, fetched):
need_parent_prefetch = False
for node_position in range(4, len(path_nodes) - 1):
redfish_path = f'/redfish/v1/{"/".join(path_nodes[3:node_position + 1])}'
logging.info(f"Checking redfish path: {redfish_path}")
logger.info(f"Checking redfish path: {redfish_path}")
if redfish_path not in visited:
need_parent_prefetch = True
logging.info(f"Inspect redfish path: {redfish_path}")
logger.info(f"Inspect redfish path: {redfish_path}")
queue.append(redfish_path)
visited.append(redfish_path)
if need_parent_prefetch: # requeue
Expand All @@ -289,7 +314,7 @@ def fetchResourceAndTree(self, id, aggregation_source, visited, queue, fetched):

def fetchResource(self, obj_id, aggregation_source):
resource_endpoint = aggregation_source["HostName"] + obj_id
logging.info(f"fetch: {resource_endpoint}")
logger.info(f"fetch: {resource_endpoint}")
response = requests.get(resource_endpoint)

if response.status_code == 200:
Expand All @@ -316,7 +341,8 @@ def createInspectedObject(self,redfish_obj, aggregation_source):
except ResourceNotFound:
add_aggregation_source_reference(redfish_obj, aggregation_source)
self.create_object(file_path, redfish_obj)

else:
logger.debug("This is a collection")

def add_aggregation_source_reference(redfish_obj, aggregation_source):
if "Oem" not in redfish_obj:
Expand Down
46 changes: 35 additions & 11 deletions sunfish/lib/object_handler.py
View file
Original file line number Diff line number Diff line change
@@ -1,15 +1,39 @@
# Copyright IBM Corp. 2023
# This software is available to you under a BSD 3-Clause License.
# The full license terms are available here: https://github.com/OpenFabrics/sunfish_library_reference/blob/main/LICENSE
import logging

import sunfish.lib.core
from sunfish.models.types import *

logger = logging.getLogger("RedfishObjectHandler")


class RedfishObjectHandlersTable:
@classmethod
def ComputerSystem(cls, core: 'sunfish.lib.core.Core', path: str, operation: SunfishRequestType, payload: dict):
return "ObjectHandler ComputerSystem"

@classmethod
def EventDestination(cls, core: 'sunfish.lib.core.Core', path: str, operation: SunfishRequestType, payload: dict):
if operation == SunfishRequestType.CREATE:
core.subscription_handler.new_subscription(payload)
elif operation == SunfishRequestType.REPLACE or operation == SunfishRequestType.PATCH:
core.subscription_handler.delete_subscription(payload)
core.subscription_handler.new_subscription(payload)
elif operation == SunfishRequestType.DELETE:
core.subscription_handler.delete_subscription(path)


class RedfishObjectHandler:
dispatch_table = {
"ComputerSystem": RedfishObjectHandlersTable.ComputerSystem,
"EventDestination": RedfishObjectHandlersTable.EventDestination
}
@classmethod
def dispatch(cls, core: 'sunfish.lib.core.Core', object_type: str, path: str,
operation: SunfishRequestType, payload: dict = None):
if object_type in cls.dispatch_table:
return cls.dispatch_table[object_type](core, path, operation, payload)
logger.debug(f"Object type '{object_type}' does not have a custom handler")

class ObjectHandler:
def __init__(self, core):
"""init that sets the conf and calls the load subcriptions method
Args:
conf (dict): dictionary where are specified the storage implementation type and the specific backend configuration.
"""
self.core = core

def ComputerSystem(self, payload):
return "ObjectHandler ComputerSystem"

0 comments on commit aee63d8

Please sign in to comment.