PIOT-CDA-02-007: Connect SystemCpuUtilTask and SystemMemUtilTask to SystemPerformanceManager

[labbenchstudios]

Description

• Create an instance of SystemCpuUtilTask and SystemMemUtilTask within SystemPerformanceManager and use the apscheduler library to run each task at a regular interval.
  • NOTE: These instructions make use of the following open source project(s):
  • The APScheduler open source library, located at: APScheduler. Reference: Reference: A. Grönholm. APScheduler. (2020) [Online]. Available: https://pypi.org/project/APScheduler/.

Review the README

• Please see README.md for further information on, and use of, this content.
• License for embedded documentation and source codes: PIOT-DOC-LIC

Estimated effort may vary greatly

• The estimated level of effort for this exercise shown in the 'Estimate' section below is a very rough approximation. The actual level of effort may vary greatly depending on your development and test environment, experience with the requisite technologies, and many other factors.

Actions

NOTE: The implementation examples depicted here are only one way to implement the requirements listed. Your own implementation may vary of course.

• Add the following import statement (you may need to add others as well to support the constructor implementation described next):

  from apscheduler.schedulers.background import BackgroundScheduler

• Update the constructor as follows:

  def __init__(self):
  configUtil = ConfigUtil()
  
  self.pollRate = \
      configUtil.getInteger( \
          section = ConfigConst.CONSTRAINED_DEVICE, key = ConfigConst.POLL_CYCLES_KEY, defaultVal = ConfigConst.DEFAULT_POLL_CYCLES)
  
  self.locationID = \
      configUtil.getProperty( \
          section = ConfigConst.CONSTRAINED_DEVICE, key = ConfigConst.DEVICE_LOCATION_ID_KEY, defaultVal = ConfigConst.NOT_SET)
  
  if self.pollRate <= 0:
      self.pollRate = ConfigConst.DEFAULT_POLL_CYCLES
  
  self.dataMsgListener = None
  
  # NOTE: The next four SLOC's are new for this task
  self.scheduler = BackgroundScheduler()
  self.scheduler.add_job(self.handleTelemetry, 'interval', seconds = self.pollRate)
  
  self.cpuUtilTask = SystemCpuUtilTask()
  self.memUtilTask = SystemMemUtilTask()

• NOTES:
  • The add_job() method requires a function pointer to the handleTelemetry() method, which is already stubbed out.
  • Be sure to read the documentation for add_job() - you may want to consider using additional parameters, such as coalesce and misfire_grace_time if you run into thread call delays or failures. See apscheduler v3.x - add_job() docs for further info. Here's one notional example for review purposes:

    self.scheduler = BackgroundScheduler()
    self.scheduler.add_job( \
    self.handleTelemetry, 'interval', seconds = self.pollRate, \
    max_instances = 2, coalesce = True, misfire_grace_time = 15)

• Create a public method named handleTelemetry() and add the following lines of code:

  • NOTE: Some implementation examples discussed during lab module reviews show cpuUtilPct and memUtilPct as class-scoped. Since these variables are only needed within the handleTelemetry() method, they can be locally scoped to the method.

    
    def handleTelemetry(self):
    cpuUtilPct = self.cpuUtilTask.getTelemetryValue()
    memUtilPct = self.memUtilTask.getTelemetryValue()

  logging.debug('CPU utilization is %s percent, and memory utilization is %s percent.', str(cpuUtilPct), str(memUtilPct))

• Implement the startManager() and stopManager() methods:

  • NOTE: the implementation for each will look similar to the following (be sure to comment your code, and feel free to change the log messages to meet your specific needs):

    
    def startManager(self):
    logging.info("Starting SystemPerformanceManager...")

  if not self.scheduler.running: self.scheduler.start() logging.info("Started SystemPerformanceManager.") else: logging.warning("SystemPerformanceManager scheduler already started. Ignoring.")

def stopManager(self): logging.info("Stopping SystemPerformanceManager...")

try:
    self.scheduler.shutdown()
    logging.info("Stopped SystemPerformanceManager.")
except:
    logging.warning("SystemPerformanceManager scheduler already stopped. Ignoring.")

**Estimate**

- Small

**Tests**

- Integration tests (in ./src/test/python/programmingtheiot/part01/integration)
  - Run ./system/SystemPerformanceManagerTest. Integration test should pass and generate output similar to the following:

Finding files... done. Importing test modules ... done.

2020-09-05 22:47:52,609:SystemPerformanceManagerTest:INFO:Testing SystemPerformanceManager class... 2020-09-05 22:47:52,761:base:INFO:Adding job tentatively -- it will be properly scheduled when the scheduler starts 2020-09-05 22:47:52,762:SystemPerformanceManager:INFO:Started SystemPerformanceManager. 2020-09-05 22:47:52,763:base:INFO:Added job "SystemPerformanceManager.handleTelemetry" to job store "default" 2020-09-05 22:47:52,763:base:INFO:Scheduler started 2020-09-05 22:47:52,764:base:DEBUG:Looking for jobs to run 2020-09-05 22:47:52,764:base:DEBUG:Next wakeup is due at 2020-09-05 22:47:57.761111-04:00 (in 4.997008 seconds) 2020-09-05 22:47:57,762:base:DEBUG:Looking for jobs to run 2020-09-05 22:47:57,763:base:INFO:Running job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], nex t run at: 2020-09-05 22:47:57 EDT)" (scheduled at 2020-09-05 22:47:57.761111-04:00) 2020-09-05 22:47:57,763:SystemPerformanceManager:DEBUG:CPU utilization is 10.0 percent, and memory utilization is 64.1 pe rcent. . . . 2020-09-05 22:48:52,762:base:INFO:Job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at : 2020-09-05 22:48:57 EDT)" executed successfully 2020-09-05 22:48:52,764:SystemPerformanceManager:INFO:Stopped SystemPerformanceManager. 2020-09-05 22:48:52,764:base:INFO:Scheduler has been shut down 2020-09-05 22:48:52,764:base:DEBUG:Looking for jobs to run 2020-09-05 22:48:52,765:base:DEBUG:No jobs; waiting until a job is added

Ran 1 test in 60.156s

OK

[MAbuShattal]

Hi, This line of code:

        self.scheduler.add_job( 'self.handleTelemetry',  'interval', seconds = pollRate)

output the following error:

Finding files... done.
Importing test modules ... done.

2021-12-06 16:09:01,256:SystemPerformanceManagerTest:INFO:Testing SystemPerformanceManager class...
2021-12-06 16:09:01,256:ConfigUtil:INFO:Loading config: ../../../../../../../config/PiotConfig.props
2021-12-06 16:09:01,256:ConfigUtil:DEBUG:Config: ['Mqtt.GatewayService', 'Coap.GatewayService', 'ConstrainedDevice']
2021-12-06 16:09:01,257:ConfigUtil:INFO:Created instance of ConfigUtil: <programmingtheiot.common.ConfigUtil.ConfigUtil object at 0x7fb64dc6a3a0>
======================================================================
ERROR: setUpClass (src.test.python.programmingtheiot.part01.integration.system.SystemPerformanceManagerTest.SystemPerformanceManagerTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/mas/programmingtheiot/python-components/src/test/python/programmingtheiot/part01/integration/system/SystemPerformanceManagerTest.py", line 43, in setUpClass
    self.spMgr = SystemPerformanceManager()
  File "/home/mas/programmingtheiot/python-components/src/main/python/programmingtheiot/cda/system/SystemPerformanceManager.py", line 46, in __init__
    self.scheduler.add_job( 'self.handleTelemetry',  'interval', seconds = pollRate)
NameError: name 'pollRate' is not defined

----------------------------------------------------------------------
Ran 0 tests in 0.007s

FAILED (errors=1)

---

---

Trying:

        self.scheduler.add_job( 'self.handleTelemetry',  'interval', seconds = self.pollRate)

Output the following error:

Finding files... done.
Importing test modules ... done.

2021-12-06 16:09:48,579:SystemPerformanceManagerTest:INFO:Testing SystemPerformanceManager class...
2021-12-06 16:09:48,580:ConfigUtil:INFO:Loading config: ../../../../../../../config/PiotConfig.props
2021-12-06 16:09:48,580:ConfigUtil:DEBUG:Config: ['Mqtt.GatewayService', 'Coap.GatewayService', 'ConstrainedDevice']
2021-12-06 16:09:48,580:ConfigUtil:INFO:Created instance of ConfigUtil: <programmingtheiot.common.ConfigUtil.ConfigUtil object at 0x7fa719ed53a0>
/usr/local/lib/python3.8/dist-packages/apscheduler/util.py:95: PytzUsageWarning: The zone attribute is specific to pytz's interface; please migrate to a new time zone provider. For more details on how to do so, see https://pytz-deprecation-shim.readthedocs.io/en/latest/migration.html
  if obj.zone == 'local':
======================================================================
ERROR: setUpClass (src.test.python.programmingtheiot.part01.integration.system.SystemPerformanceManagerTest.SystemPerformanceManagerTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/mas/programmingtheiot/python-components/src/test/python/programmingtheiot/part01/integration/system/SystemPerformanceManagerTest.py", line 43, in setUpClass
    self.spMgr = SystemPerformanceManager()
  File "/home/mas/programmingtheiot/python-components/src/main/python/programmingtheiot/cda/system/SystemPerformanceManager.py", line 46, in __init__
    self.scheduler.add_job( 'self.handleTelemetry',  'interval', seconds = self.pollRate)
  File "/usr/local/lib/python3.8/dist-packages/apscheduler/schedulers/base.py", line 439, in add_job
    job = Job(self, **job_kwargs)
  File "/usr/local/lib/python3.8/dist-packages/apscheduler/job.py", line 49, in __init__
    self._modify(id=id or uuid4().hex, **kwargs)
  File "/usr/local/lib/python3.8/dist-packages/apscheduler/job.py", line 162, in _modify
    func = ref_to_obj(func)
  File "/usr/local/lib/python3.8/dist-packages/apscheduler/util.py", line 300, in ref_to_obj
    raise ValueError('Invalid reference')
ValueError: Invalid reference

----------------------------------------------------------------------
Ran 0 tests in 0.039s

FAILED (errors=1)

Any thought?

Best, Mohammad Abu Shattal

Full code of SystemPerformanceManager:

#####
# 
# This class is part of the Programming the Internet of Things project.
# 
# It is provided as a simple shell to guide the student and assist with
# implementation for the Programming the Internet of Things exercises,
# and designed to be modified by the student as needed.
#

import logging

from apscheduler.schedulers.background import BackgroundScheduler

import programmingtheiot.common.ConfigConst as ConfigConst

from programmingtheiot.common.ConfigUtil import ConfigUtil
from programmingtheiot.common.IDataMessageListener import IDataMessageListener

from programmingtheiot.cda.system.SystemCpuUtilTask import SystemCpuUtilTask
from programmingtheiot.cda.system.SystemMemUtilTask import SystemMemUtilTask

from programmingtheiot.data.SystemPerformanceData import SystemPerformanceData

class SystemPerformanceManager(object):
    """
    Shell representation of class for student implementation.

    """

    def __init__(self):
        self.configUtil = ConfigUtil()
        self.pollRate =  self.configUtil.getInteger( 
             section = ConfigConst.CONSTRAINED_DEVICE, 
              key = ConfigConst.POLL_CYCLES_KEY, 
               defaultVal = ConfigConst.DEFAULT_POLL_CYCLES)

        self.locationID =  self.configUtil.getProperty( 
             section = ConfigConst.CONSTRAINED_DEVICE, 
              key = ConfigConst.DEVICE_LOCATION_ID_KEY, 
               defaultVal = ConfigConst.CONSTRAINED_DEVICE)
        if self.pollRate <= 0: 
            self.pollRate = ConfigConst.DEFAULT_POLL_CYCLES

        self.scheduler = BackgroundScheduler() 
        self.scheduler.add_job( 'self.handleTelemetry',  'interval', seconds = self.pollRate)

        self.dataMsgListener = None

    def handleTelemetry(self):
        self.cpuUtilTask = SystemCpuUtilTask() 
        self.memUtilTask = SystemMemUtilTask()
        logging.debug('CPU utilization is %s percent, and memory utilization is %s percent.', str(self.cpuUtilPct), str(self.memUtilPct))

    def setDataMessageListener(self, listener: IDataMessageListener) -> bool:
        pass

    def startManager(self):
        logging.info("Started SystemPerformanceManager")
        if not self.scheduler.running: 
            self.scheduler.start() 
        else: 
            logging.warning(  "SystemPerformanceManager scheduler already started. Ignoring.")

    def stopManager(self):
        logging.info("Stopped SystemPerformanceManager")
        try: 
            self.scheduler.shutdown()
        except: 
            logging.warning(  "SystemPerformanceManager scheduler already stopped. Ignoring.")

[labbenchstudios]

Hi - thanks for pointing this out and for your question. Two things:

1. In the book, the code on p. 61 (Part I, Chapter 2) has two errors, which I've corrected on the errata page here: Programming the IoT - Errata and Clarifications. The code samples under 'Actions' reflect these corrections.
2. You may want to consider creating the instance of SystemCpuUtilTask() and SystemMemUtilTask() in the constructor instead, and just call the getTelemetryValue() function on each instance from within handleTelemetry() (see the 2nd and 3rd bullets under 'Actions' - I added some clarifications that I hope will be useful).

Hope this helps!

[MAbuShattal]

Thank you for your quick help! That helped but I needed to update the code to include definitions for cpuUtilPct and memUtilPct to be members of the SystemPerformanceManager Class. Below is the full code of SystemPerformance that worked correctly:

#####
# 
# This class is part of the Programming the Internet of Things project.
# 
# It is provided as a simple shell to guide the student and assist with
# implementation for the Programming the Internet of Things exercises,
# and designed to be modified by the student as needed.
#

import logging

from apscheduler.schedulers.background import BackgroundScheduler

import programmingtheiot.common.ConfigConst as ConfigConst

from programmingtheiot.common.ConfigUtil import ConfigUtil
from programmingtheiot.common.IDataMessageListener import IDataMessageListener

from programmingtheiot.cda.system.SystemCpuUtilTask import SystemCpuUtilTask
from programmingtheiot.cda.system.SystemMemUtilTask import SystemMemUtilTask

from programmingtheiot.data.SystemPerformanceData import SystemPerformanceData

class SystemPerformanceManager(object):
    """
    Shell representation of class for student implementation.

    """

    def __init__(self):
        self.configUtil = ConfigUtil()
        self.pollRate =  self.configUtil.getInteger( 
             section = ConfigConst.CONSTRAINED_DEVICE, 
              key = ConfigConst.POLL_CYCLES_KEY, 
               defaultVal = ConfigConst.DEFAULT_POLL_CYCLES)

        self.locationID =  self.configUtil.getProperty( 
             section = ConfigConst.CONSTRAINED_DEVICE, 
              key = ConfigConst.DEVICE_LOCATION_ID_KEY, 
               defaultVal = ConfigConst.CONSTRAINED_DEVICE)
        if self.pollRate <= 0: 
            self.pollRate = ConfigConst.DEFAULT_POLL_CYCLES

        self.scheduler = BackgroundScheduler() 
        self.scheduler.add_job( self.handleTelemetry,  'interval', seconds = self.pollRate)

        self.cpuUtilTask = SystemCpuUtilTask() 
        self.memUtilTask = SystemMemUtilTask()

        self.cpuUtilPct = self.cpuUtilTask.getTelemetryValue()
        self.memUtilPct = self.memUtilTask.getTelemetryValue()

        self.dataMsgListener = None

    def handleTelemetry(self):
        cpuUtilPct = self.cpuUtilTask.getTelemetryValue()
        memUtilPct = self.memUtilTask.getTelemetryValue()

        logging.debug('CPU utilization is %s percent, and memory utilization is %s percent.', str(self.cpuUtilPct), str(self.memUtilPct))

    def setDataMessageListener(self, listener: IDataMessageListener) -> bool:
        pass

    def startManager(self):
        logging.info("Started SystemPerformanceManager")
        if not self.scheduler.running: 
            self.scheduler.start() 
        else: 
            logging.warning(  "SystemPerformanceManager scheduler already started. Ignoring.")

    def stopManager(self):
        logging.info("Stopped SystemPerformanceManager")
        try: 
            self.scheduler.shutdown()
        except: 
            logging.warning(  "SystemPerformanceManager scheduler already stopped. Ignoring.")

Below are the results that I have for this code. I see the results are the same during one run (multiple iterations over time. e.g.: 85.7 in this run). When I run the code it gives a new value (e.g. 98.2 )but it is repeated during the same run.

Finding files... done.
Importing test modules ... done.

2021-12-07 12:23:10,371:SystemPerformanceManagerTest:INFO:Testing SystemPerformanceManager class...
2021-12-07 12:23:10,371:ConfigUtil:INFO:Loading config: ../../../../../../../config/PiotConfig.props
2021-12-07 12:23:10,372:ConfigUtil:DEBUG:Config: ['Mqtt.GatewayService', 'Coap.GatewayService', 'ConstrainedDevice']
2021-12-07 12:23:10,372:ConfigUtil:INFO:Created instance of ConfigUtil: <programmingtheiot.common.ConfigUtil.ConfigUtil object at 0x7fa2c5dce3a0>
/usr/local/lib/python3.8/dist-packages/apscheduler/util.py:95: PytzUsageWarning: The zone attribute is specific to pytz's interface; please migrate to a new time zone provider. For more details on how to do so, see https://pytz-deprecation-shim.readthedocs.io/en/latest/migration.html
  if obj.zone == 'local':
2021-12-07 12:23:10,414:base:INFO:Adding job tentatively -- it will be properly scheduled when the scheduler starts
2021-12-07 12:23:10,414:SystemPerformanceManager:INFO:Started SystemPerformanceManager
/usr/local/lib/python3.8/dist-packages/apscheduler/triggers/interval.py:66: PytzUsageWarning: The normalize method is no longer necessary, as this time zone supports the fold attribute (PEP 495). For more details on migrating to a PEP 495-compliant implementation, see https://pytz-deprecation-shim.readthedocs.io/en/latest/migration.html
  return self.timezone.normalize(next_fire_time)
2021-12-07 12:23:10,419:base:INFO:Added job "SystemPerformanceManager.handleTelemetry" to job store "default"
2021-12-07 12:23:10,419:base:INFO:Scheduler started
2021-12-07 12:23:10,419:base:DEBUG:Looking for jobs to run
2021-12-07 12:23:10,419:base:DEBUG:Next wakeup is due at 2021-12-07 12:23:15.413896-05:00 (in 4.993956 seconds)
2021-12-07 12:23:15,414:base:DEBUG:Looking for jobs to run
2021-12-07 12:23:15,414:base:INFO:Running job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at: 2021-12-07 12:23:15 EST)" (scheduled at 2021-12-07 12:23:15.413896-05:00)
2021-12-07 12:23:15,415:base:DEBUG:Next wakeup is due at 2021-12-07 12:23:20.413896-05:00 (in 4.999609 seconds)
2021-12-07 12:23:15,415:SystemPerformanceManager:DEBUG:CPU utilization is 85.7 percent, and memory utilization is None percent.
2021-12-07 12:23:15,415:base:INFO:Job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at: 2021-12-07 12:23:20 EST)" executed successfully
2021-12-07 12:23:20,414:base:DEBUG:Looking for jobs to run
2021-12-07 12:23:20,415:base:DEBUG:Next wakeup is due at 2021-12-07 12:23:25.413896-05:00 (in 4.998919 seconds)
2021-12-07 12:23:20,415:base:INFO:Running job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at: 2021-12-07 12:23:25 EST)" (scheduled at 2021-12-07 12:23:20.413896-05:00)
2021-12-07 12:23:20,415:SystemPerformanceManager:DEBUG:CPU utilization is 85.7 percent, and memory utilization is None percent.
2021-12-07 12:23:20,415:base:INFO:Job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at: 2021-12-07 12:23:25 EST)" executed successfully
2021-12-07 12:23:25,414:base:DEBUG:Looking for jobs to run
2021-12-07 12:23:25,414:base:DEBUG:Next wakeup is due at 2021-12-07 12:23:30.413896-05:00 (in 4.999570 seconds)
2021-12-07 12:23:25,414:base:INFO:Running job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at: 2021-12-07 12:23:30 EST)" (scheduled at 2021-12-07 12:23:25.413896-05:00)
2021-12-07 12:23:25,414:SystemPerformanceManager:DEBUG:CPU utilization is 85.7 percent, and memory utilization is None percent.
2021-12-07 12:23:25,414:base:INFO:Job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at: 2021-12-07 12:23:30 EST)" executed successfully
2021-12-07 12:23:30,414:base:DEBUG:Looking for jobs to run
2021-12-07 12:23:30,414:base:DEBUG:Next wakeup is due at 2021-12-07 12:23:35.413896-05:00 (in 4.999605 seconds)
2021-12-07 12:23:30,414:base:INFO:Running job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at: 2021-12-07 12:23:35 EST)" (scheduled at 2021-12-07 12:23:30.413896-05:00)
2021-12-07 12:23:30,414:SystemPerformanceManager:DEBUG:CPU utilization is 85.7 percent, and memory utilization is None percent.
2021-12-07 12:23:30,414:base:INFO:Job "SystemPerformanceManager.handleTelemetry (trigger: interval[0:00:05], next run at: 2021-12-07 12:23:35 EST)" executed successfully

[labbenchstudios]

Glad to hear the fixes mentioned resolved the initial issue. Here are some thoughts on your follow-up question:

• The generateTelemetry() function should be the only place you need to invoke the getTelemetryValue() function on each task instance, as it will be invoked during each poll cycle from the scheduler.
• You may also want to to double check the task implementations specified in PIOT-CDA-02-005 and PIOT-CDA-02-006. If, for instance, you use the method-scoped cpuUtilPct to store the value from the task in handleTelemetry(), be sure to use the same reference in any logging you may want to include. The example shows a class-scoped self.cpuUtilPct - if you go with that approach, use self.cpuUtilPct to retrieve the value.
• Once you verify the scoping of the variable is correct, keep in mind you may not see much of a delta in each value from poll to poll. One way to check this is to start another application while running the test above to see if both memory and CPU utilization increase (they probably will - at least briefly).

Hope this helps!