We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix import error
```
>>> from parsl.dataflow.states import FINAL_FAILED_STATES
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/annawoodard/parsl/parsl/__init__.py", line 37, in <module>
    from parsl.dataflow.dflow import DataFlowKernel, DataFlowKernelLoader
  File "/home/annawoodard/parsl/parsl/dataflow/dflow.py", line 31, in <module>
    from parsl.dataflow.usage_tracking.usage import UsageTracker
  File "/home/annawoodard/parsl/parsl/dataflow/usage_tracking/usage.py", line 13, in <module>
    from parsl.dataflow.states import FINAL_FAILED_STATES
ImportError: cannot import name 'FINAL_FAILED_STATES'
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `parsl/dataflow/usage_tracking/usage.py`
Content:
```
1 import uuid
2 import time
3 import hashlib
4 import os
5 import getpass
6 import json
7 import logging
8 import socket
9 import sys
10 import platform
11 import multiprocessing as mp
12 
13 from parsl.dataflow.states import FINAL_FAILED_STATES
14 from parsl.version import VERSION as PARSL_VERSION
15 
16 logger = logging.getLogger(__name__)
17 
18 
19 def async_process(fn):
20     """ Decorator function to launch a function as a separate process """
21 
22     def run(*args, **kwargs):
23         proc = mp.Process(target=fn, args=args, kwargs=kwargs)
24         proc.start()
25         return proc
26 
27     return run
28 
29 
30 @async_process
31 def udp_messenger(domain_name, UDP_IP, UDP_PORT, sock_timeout, message):
32     """Send UDP messages to usage tracker asynchronously
33 
34     This multiprocessing based messenger was written to overcome the limitations
35     of signalling/terminating a thread that is blocked on a system call. This
36     messenger is created as a separate process, and initialized with 2 queues,
37     to_send to receive messages to be sent to the internet.
38 
39     Args:
40           - domain_name (str) : Domain name string
41           - UDP_IP (str) : IP address YYY.YYY.YYY.YYY
42           - UDP_PORT (int) : UDP port to send out on
43           - sock_timeout (int) : Socket timeout
44           - to_send (multiprocessing.Queue) : Queue of outgoing messages to internet
45     """
46     try:
47         if message is None:
48             raise ValueError("message was none")
49 
50         encoded_message = bytes(message, "utf-8")
51 
52         if encoded_message is None:
53             raise ValueError("utf-8 encoding of message failed")
54 
55         if domain_name:
56             try:
57                 UDP_IP = socket.gethostbyname(domain_name)
58             except Exception:
59                 # (False, "Domain lookup failed, defaulting to {0}".format(UDP_IP))
60                 pass
61 
62         if UDP_IP is None:
63             raise Exception("UDP_IP is None")
64 
65         if UDP_PORT is None:
66             raise Exception("UDP_PORT is None")
67 
68         sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # UDP
69         sock.settimeout(sock_timeout)
70         sock.sendto(bytes(message, "utf-8"), (UDP_IP, UDP_PORT))
71         sock.close()
72 
73     except socket.timeout:
74         logger.debug("Failed to send usage tracking data: socket timeout")
75     except OSError as e:
76         logger.debug("Failed to send usage tracking data: OSError: {}".format(e))
77     except Exception as e:
78         logger.debug("Failed to send usage tracking data: Exception: {}".format(e))
79 
80 
81 class UsageTracker (object):
82     """Anonymized Usage Tracking for Parsl.
83 
84     Client for this is here : https://github.com/Parsl/parsl_tracking
85     This issue captures the discussion that went into functionality
86     implemented here : https://github.com/Parsl/parsl/issues/34
87 
88     """
89 
90     def __init__(self, dfk, ip='52.3.111.203', port=50077,
91                  domain_name='tracking.parsl-project.org'):
92         """Initialize usage tracking unless the user has opted-out.
93 
94         We will try to resolve the hostname specified in kwarg:domain_name
95         and if that fails attempt to use the kwarg:ip. Determining the
96         IP and sending message is threaded to avoid slowing down DFK
97         initialization.
98 
99         Tracks usage stats by inspecting the internal state of the dfk.
100 
101         Args:
102              - dfk (DFK object) : Data Flow Kernel object
103 
104         KWargs:
105              - ip (string) : IP address
106              - port (int) : Port number, Default:50077
107              - domain_name (string) : Domain name, will override IP
108                   Default: tracking.parsl-project.org
109         """
110 
111         self.domain_name = domain_name
112         self.ip = ip
113         # The sock timeout will only apply to UDP send and not domain resolution
114         self.sock_timeout = 5
115         self.UDP_PORT = port
116         self.UDP_IP = None
117         self.procs = []
118         self.dfk = dfk
119         self.config = self.dfk.config
120         self.uuid = str(uuid.uuid4())
121         self.parsl_version = PARSL_VERSION
122         self.python_version = "{}.{}.{}".format(sys.version_info.major,
123                                                 sys.version_info.minor,
124                                                 sys.version_info.micro)
125         self.test_mode, self.tracking_enabled = self.check_tracking_enabled()
126         logger.debug("Tracking status: {}".format(self.tracking_enabled))
127         logger.debug("Testing mode   : {}".format(self.test_mode))
128         self.initialized = False  # Once first message is sent this will be True
129 
130     def check_tracking_enabled(self):
131         """By default tracking is enabled.
132 
133         If Test mode is set via env variable PARSL_TESTING, a test flag is set
134 
135         Tracking is disabled if :
136             1. config["globals"]["usageTracking"] is set to False (Bool)
137             2. Environment variable PARSL_TRACKING is set to false (case insensitive)
138 
139         """
140         track = True   # By default we track usage
141         test = False  # By default we are not in testing mode
142 
143         testvar = str(os.environ.get("PARSL_TESTING", 'None')).lower()
144         if testvar == 'true':
145             test = True
146 
147         if not self.config.usage_tracking:
148             track = False
149 
150         envvar = str(os.environ.get("PARSL_TRACKING", True)).lower()
151         if envvar == "false":
152             track = False
153 
154         return test, track
155 
156     def construct_start_message(self):
157         """Collect preliminary run info at the start of the DFK.
158 
159         Returns :
160               - Message dict dumped as json string, ready for UDP
161         """
162         uname = getpass.getuser().encode('latin1')
163         hashed_username = hashlib.sha256(uname).hexdigest()[0:10]
164         hname = socket.gethostname().encode('latin1')
165         hashed_hostname = hashlib.sha256(hname).hexdigest()[0:10]
166         message = {'uuid': self.uuid,
167                    'uname': hashed_username,
168                    'hname': hashed_hostname,
169                    'test': self.test_mode,
170                    'parsl_v': self.parsl_version,
171                    'python_v': self.python_version,
172                    'os': platform.system(),
173                    'os_v': platform.release(),
174                    'start': time.time()}
175 
176         return json.dumps(message)
177 
178     def construct_end_message(self):
179         """Collect the final run information at the time of DFK cleanup.
180 
181         Returns:
182              - Message dict dumped as json string, ready for UDP
183         """
184         app_count = self.dfk.task_count
185 
186         site_count = len([x for x in self.dfk.config.executors if x.managed])
187 
188         failed_states = FINAL_FAILED_STATES
189         app_fails = len([t for t in self.dfk.tasks if
190                          self.dfk.tasks[t]['status'] in failed_states])
191 
192         message = {'uuid': self.uuid,
193                    'end': time.time(),
194                    't_apps': app_count,
195                    'sites': site_count,
196                    'c_time': None,
197                    'failed': app_fails,
198                    'test': self.test_mode,
199                    }
200 
201         return json.dumps(message)
202 
203     def send_UDP_message(self, message):
204         """Send UDP message."""
205         x = 0
206         if self.tracking_enabled:
207             try:
208                 proc = udp_messenger(self.domain_name, self.UDP_IP, self.UDP_PORT, self.sock_timeout, message)
209                 self.procs.append(proc)
210             except Exception as e:
211                 logger.debug("Usage tracking failed: {}".format(e))
212         else:
213             x = -1
214 
215         return x
216 
217     def send_message(self):
218         """Send message over UDP.
219 
220         If tracking is disables, the bytes_sent will always be set to -1
221 
222         Returns:
223             (bytes_sent, time_taken)
224         """
225         start = time.time()
226         message = None
227         if not self.initialized:
228             message = self.construct_start_message()
229             self.initialized = True
230         else:
231             message = self.construct_end_message()
232 
233         self.send_UDP_message(message)
234         end = time.time()
235 
236         return end - start
237 
238     def __del__(self):
239         return self.close()
240 
241     def close(self):
242         """We terminate (SIGTERM) the processes added to the self.procs list """
243         for proc in self.procs:
244             proc.terminate()
245 
246 
247 if __name__ == '__main__':
248 
249     from parsl import *
250 
251     set_stream_logger()
252     workers = ThreadPoolExecutor(max_workers=4)
253     dfk = DataFlowKernel(executors=[workers])
254 
255     dfk.cleanup()
256
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/parsl/dataflow/usage_tracking/usage.py b/parsl/dataflow/usage_tracking/usage.py
--- a/parsl/dataflow/usage_tracking/usage.py
+++ b/parsl/dataflow/usage_tracking/usage.py
@@ -10,7 +10,7 @@
 import platform
 import multiprocessing as mp
 
-from parsl.dataflow.states import FINAL_FAILED_STATES
+from parsl.dataflow.states import FINAL_FAILURE_STATES
 from parsl.version import VERSION as PARSL_VERSION
 
 logger = logging.getLogger(__name__)
@@ -185,9 +185,8 @@
 
         site_count = len([x for x in self.dfk.config.executors if x.managed])
 
-        failed_states = FINAL_FAILED_STATES
         app_fails = len([t for t in self.dfk.tasks if
-                         self.dfk.tasks[t]['status'] in failed_states])
+                         self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])
 
         message = {'uuid': self.uuid,
                    'end': time.time(),

{"golden_diff": "diff --git a/parsl/dataflow/usage_tracking/usage.py b/parsl/dataflow/usage_tracking/usage.py\n--- a/parsl/dataflow/usage_tracking/usage.py\n+++ b/parsl/dataflow/usage_tracking/usage.py\n@@ -10,7 +10,7 @@\n import platform\n import multiprocessing as mp\n \n-from parsl.dataflow.states import FINAL_FAILED_STATES\n+from parsl.dataflow.states import FINAL_FAILURE_STATES\n from parsl.version import VERSION as PARSL_VERSION\n \n logger = logging.getLogger(__name__)\n@@ -185,9 +185,8 @@\n \n         site_count = len([x for x in self.dfk.config.executors if x.managed])\n \n-        failed_states = FINAL_FAILED_STATES\n         app_fails = len([t for t in self.dfk.tasks if\n-                         self.dfk.tasks[t]['status'] in failed_states])\n+                         self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])\n \n         message = {'uuid': self.uuid,\n                    'end': time.time(),\n", "issue": "Fix import error\n```\r\n>>> from parsl.dataflow.states import FINAL_FAILED_STATES\r\nTraceback (most recent call last):\r\n  File \"<stdin>\", line 1, in <module>\r\n  File \"/home/annawoodard/parsl/parsl/__init__.py\", line 37, in <module>\r\n    from parsl.dataflow.dflow import DataFlowKernel, DataFlowKernelLoader\r\n  File \"/home/annawoodard/parsl/parsl/dataflow/dflow.py\", line 31, in <module>\r\n    from parsl.dataflow.usage_tracking.usage import UsageTracker\r\n  File \"/home/annawoodard/parsl/parsl/dataflow/usage_tracking/usage.py\", line 13, in <module>\r\n    from parsl.dataflow.states import FINAL_FAILED_STATES\r\nImportError: cannot import name 'FINAL_FAILED_STATES'\r\n```\n", "before_files": [{"content": "import uuid\nimport time\nimport hashlib\nimport os\nimport getpass\nimport json\nimport logging\nimport socket\nimport sys\nimport platform\nimport multiprocessing as mp\n\nfrom parsl.dataflow.states import FINAL_FAILED_STATES\nfrom parsl.version import VERSION as PARSL_VERSION\n\nlogger = logging.getLogger(__name__)\n\n\ndef async_process(fn):\n    \"\"\" Decorator function to launch a function as a separate process \"\"\"\n\n    def run(*args, **kwargs):\n        proc = mp.Process(target=fn, args=args, kwargs=kwargs)\n        proc.start()\n        return proc\n\n    return run\n\n\n@async_process\ndef udp_messenger(domain_name, UDP_IP, UDP_PORT, sock_timeout, message):\n    \"\"\"Send UDP messages to usage tracker asynchronously\n\n    This multiprocessing based messenger was written to overcome the limitations\n    of signalling/terminating a thread that is blocked on a system call. This\n    messenger is created as a separate process, and initialized with 2 queues,\n    to_send to receive messages to be sent to the internet.\n\n    Args:\n          - domain_name (str) : Domain name string\n          - UDP_IP (str) : IP address YYY.YYY.YYY.YYY\n          - UDP_PORT (int) : UDP port to send out on\n          - sock_timeout (int) : Socket timeout\n          - to_send (multiprocessing.Queue) : Queue of outgoing messages to internet\n    \"\"\"\n    try:\n        if message is None:\n            raise ValueError(\"message was none\")\n\n        encoded_message = bytes(message, \"utf-8\")\n\n        if encoded_message is None:\n            raise ValueError(\"utf-8 encoding of message failed\")\n\n        if domain_name:\n            try:\n                UDP_IP = socket.gethostbyname(domain_name)\n            except Exception:\n                # (False, \"Domain lookup failed, defaulting to {0}\".format(UDP_IP))\n                pass\n\n        if UDP_IP is None:\n            raise Exception(\"UDP_IP is None\")\n\n        if UDP_PORT is None:\n            raise Exception(\"UDP_PORT is None\")\n\n        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # UDP\n        sock.settimeout(sock_timeout)\n        sock.sendto(bytes(message, \"utf-8\"), (UDP_IP, UDP_PORT))\n        sock.close()\n\n    except socket.timeout:\n        logger.debug(\"Failed to send usage tracking data: socket timeout\")\n    except OSError as e:\n        logger.debug(\"Failed to send usage tracking data: OSError: {}\".format(e))\n    except Exception as e:\n        logger.debug(\"Failed to send usage tracking data: Exception: {}\".format(e))\n\n\nclass UsageTracker (object):\n    \"\"\"Anonymized Usage Tracking for Parsl.\n\n    Client for this is here : https://github.com/Parsl/parsl_tracking\n    This issue captures the discussion that went into functionality\n    implemented here : https://github.com/Parsl/parsl/issues/34\n\n    \"\"\"\n\n    def __init__(self, dfk, ip='52.3.111.203', port=50077,\n                 domain_name='tracking.parsl-project.org'):\n        \"\"\"Initialize usage tracking unless the user has opted-out.\n\n        We will try to resolve the hostname specified in kwarg:domain_name\n        and if that fails attempt to use the kwarg:ip. Determining the\n        IP and sending message is threaded to avoid slowing down DFK\n        initialization.\n\n        Tracks usage stats by inspecting the internal state of the dfk.\n\n        Args:\n             - dfk (DFK object) : Data Flow Kernel object\n\n        KWargs:\n             - ip (string) : IP address\n             - port (int) : Port number, Default:50077\n             - domain_name (string) : Domain name, will override IP\n                  Default: tracking.parsl-project.org\n        \"\"\"\n\n        self.domain_name = domain_name\n        self.ip = ip\n        # The sock timeout will only apply to UDP send and not domain resolution\n        self.sock_timeout = 5\n        self.UDP_PORT = port\n        self.UDP_IP = None\n        self.procs = []\n        self.dfk = dfk\n        self.config = self.dfk.config\n        self.uuid = str(uuid.uuid4())\n        self.parsl_version = PARSL_VERSION\n        self.python_version = \"{}.{}.{}\".format(sys.version_info.major,\n                                                sys.version_info.minor,\n                                                sys.version_info.micro)\n        self.test_mode, self.tracking_enabled = self.check_tracking_enabled()\n        logger.debug(\"Tracking status: {}\".format(self.tracking_enabled))\n        logger.debug(\"Testing mode   : {}\".format(self.test_mode))\n        self.initialized = False  # Once first message is sent this will be True\n\n    def check_tracking_enabled(self):\n        \"\"\"By default tracking is enabled.\n\n        If Test mode is set via env variable PARSL_TESTING, a test flag is set\n\n        Tracking is disabled if :\n            1. config[\"globals\"][\"usageTracking\"] is set to False (Bool)\n            2. Environment variable PARSL_TRACKING is set to false (case insensitive)\n\n        \"\"\"\n        track = True   # By default we track usage\n        test = False  # By default we are not in testing mode\n\n        testvar = str(os.environ.get(\"PARSL_TESTING\", 'None')).lower()\n        if testvar == 'true':\n            test = True\n\n        if not self.config.usage_tracking:\n            track = False\n\n        envvar = str(os.environ.get(\"PARSL_TRACKING\", True)).lower()\n        if envvar == \"false\":\n            track = False\n\n        return test, track\n\n    def construct_start_message(self):\n        \"\"\"Collect preliminary run info at the start of the DFK.\n\n        Returns :\n              - Message dict dumped as json string, ready for UDP\n        \"\"\"\n        uname = getpass.getuser().encode('latin1')\n        hashed_username = hashlib.sha256(uname).hexdigest()[0:10]\n        hname = socket.gethostname().encode('latin1')\n        hashed_hostname = hashlib.sha256(hname).hexdigest()[0:10]\n        message = {'uuid': self.uuid,\n                   'uname': hashed_username,\n                   'hname': hashed_hostname,\n                   'test': self.test_mode,\n                   'parsl_v': self.parsl_version,\n                   'python_v': self.python_version,\n                   'os': platform.system(),\n                   'os_v': platform.release(),\n                   'start': time.time()}\n\n        return json.dumps(message)\n\n    def construct_end_message(self):\n        \"\"\"Collect the final run information at the time of DFK cleanup.\n\n        Returns:\n             - Message dict dumped as json string, ready for UDP\n        \"\"\"\n        app_count = self.dfk.task_count\n\n        site_count = len([x for x in self.dfk.config.executors if x.managed])\n\n        failed_states = FINAL_FAILED_STATES\n        app_fails = len([t for t in self.dfk.tasks if\n                         self.dfk.tasks[t]['status'] in failed_states])\n\n        message = {'uuid': self.uuid,\n                   'end': time.time(),\n                   't_apps': app_count,\n                   'sites': site_count,\n                   'c_time': None,\n                   'failed': app_fails,\n                   'test': self.test_mode,\n                   }\n\n        return json.dumps(message)\n\n    def send_UDP_message(self, message):\n        \"\"\"Send UDP message.\"\"\"\n        x = 0\n        if self.tracking_enabled:\n            try:\n                proc = udp_messenger(self.domain_name, self.UDP_IP, self.UDP_PORT, self.sock_timeout, message)\n                self.procs.append(proc)\n            except Exception as e:\n                logger.debug(\"Usage tracking failed: {}\".format(e))\n        else:\n            x = -1\n\n        return x\n\n    def send_message(self):\n        \"\"\"Send message over UDP.\n\n        If tracking is disables, the bytes_sent will always be set to -1\n\n        Returns:\n            (bytes_sent, time_taken)\n        \"\"\"\n        start = time.time()\n        message = None\n        if not self.initialized:\n            message = self.construct_start_message()\n            self.initialized = True\n        else:\n            message = self.construct_end_message()\n\n        self.send_UDP_message(message)\n        end = time.time()\n\n        return end - start\n\n    def __del__(self):\n        return self.close()\n\n    def close(self):\n        \"\"\"We terminate (SIGTERM) the processes added to the self.procs list \"\"\"\n        for proc in self.procs:\n            proc.terminate()\n\n\nif __name__ == '__main__':\n\n    from parsl import *\n\n    set_stream_logger()\n    workers = ThreadPoolExecutor(max_workers=4)\n    dfk = DataFlowKernel(executors=[workers])\n\n    dfk.cleanup()\n", "path": "parsl/dataflow/usage_tracking/usage.py"}], "after_files": [{"content": "import uuid\nimport time\nimport hashlib\nimport os\nimport getpass\nimport json\nimport logging\nimport socket\nimport sys\nimport platform\nimport multiprocessing as mp\n\nfrom parsl.dataflow.states import FINAL_FAILURE_STATES\nfrom parsl.version import VERSION as PARSL_VERSION\n\nlogger = logging.getLogger(__name__)\n\n\ndef async_process(fn):\n    \"\"\" Decorator function to launch a function as a separate process \"\"\"\n\n    def run(*args, **kwargs):\n        proc = mp.Process(target=fn, args=args, kwargs=kwargs)\n        proc.start()\n        return proc\n\n    return run\n\n\n@async_process\ndef udp_messenger(domain_name, UDP_IP, UDP_PORT, sock_timeout, message):\n    \"\"\"Send UDP messages to usage tracker asynchronously\n\n    This multiprocessing based messenger was written to overcome the limitations\n    of signalling/terminating a thread that is blocked on a system call. This\n    messenger is created as a separate process, and initialized with 2 queues,\n    to_send to receive messages to be sent to the internet.\n\n    Args:\n          - domain_name (str) : Domain name string\n          - UDP_IP (str) : IP address YYY.YYY.YYY.YYY\n          - UDP_PORT (int) : UDP port to send out on\n          - sock_timeout (int) : Socket timeout\n          - to_send (multiprocessing.Queue) : Queue of outgoing messages to internet\n    \"\"\"\n    try:\n        if message is None:\n            raise ValueError(\"message was none\")\n\n        encoded_message = bytes(message, \"utf-8\")\n\n        if encoded_message is None:\n            raise ValueError(\"utf-8 encoding of message failed\")\n\n        if domain_name:\n            try:\n                UDP_IP = socket.gethostbyname(domain_name)\n            except Exception:\n                # (False, \"Domain lookup failed, defaulting to {0}\".format(UDP_IP))\n                pass\n\n        if UDP_IP is None:\n            raise Exception(\"UDP_IP is None\")\n\n        if UDP_PORT is None:\n            raise Exception(\"UDP_PORT is None\")\n\n        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)  # UDP\n        sock.settimeout(sock_timeout)\n        sock.sendto(bytes(message, \"utf-8\"), (UDP_IP, UDP_PORT))\n        sock.close()\n\n    except socket.timeout:\n        logger.debug(\"Failed to send usage tracking data: socket timeout\")\n    except OSError as e:\n        logger.debug(\"Failed to send usage tracking data: OSError: {}\".format(e))\n    except Exception as e:\n        logger.debug(\"Failed to send usage tracking data: Exception: {}\".format(e))\n\n\nclass UsageTracker (object):\n    \"\"\"Anonymized Usage Tracking for Parsl.\n\n    Client for this is here : https://github.com/Parsl/parsl_tracking\n    This issue captures the discussion that went into functionality\n    implemented here : https://github.com/Parsl/parsl/issues/34\n\n    \"\"\"\n\n    def __init__(self, dfk, ip='52.3.111.203', port=50077,\n                 domain_name='tracking.parsl-project.org'):\n        \"\"\"Initialize usage tracking unless the user has opted-out.\n\n        We will try to resolve the hostname specified in kwarg:domain_name\n        and if that fails attempt to use the kwarg:ip. Determining the\n        IP and sending message is threaded to avoid slowing down DFK\n        initialization.\n\n        Tracks usage stats by inspecting the internal state of the dfk.\n\n        Args:\n             - dfk (DFK object) : Data Flow Kernel object\n\n        KWargs:\n             - ip (string) : IP address\n             - port (int) : Port number, Default:50077\n             - domain_name (string) : Domain name, will override IP\n                  Default: tracking.parsl-project.org\n        \"\"\"\n\n        self.domain_name = domain_name\n        self.ip = ip\n        # The sock timeout will only apply to UDP send and not domain resolution\n        self.sock_timeout = 5\n        self.UDP_PORT = port\n        self.UDP_IP = None\n        self.procs = []\n        self.dfk = dfk\n        self.config = self.dfk.config\n        self.uuid = str(uuid.uuid4())\n        self.parsl_version = PARSL_VERSION\n        self.python_version = \"{}.{}.{}\".format(sys.version_info.major,\n                                                sys.version_info.minor,\n                                                sys.version_info.micro)\n        self.test_mode, self.tracking_enabled = self.check_tracking_enabled()\n        logger.debug(\"Tracking status: {}\".format(self.tracking_enabled))\n        logger.debug(\"Testing mode   : {}\".format(self.test_mode))\n        self.initialized = False  # Once first message is sent this will be True\n\n    def check_tracking_enabled(self):\n        \"\"\"By default tracking is enabled.\n\n        If Test mode is set via env variable PARSL_TESTING, a test flag is set\n\n        Tracking is disabled if :\n            1. config[\"globals\"][\"usageTracking\"] is set to False (Bool)\n            2. Environment variable PARSL_TRACKING is set to false (case insensitive)\n\n        \"\"\"\n        track = True   # By default we track usage\n        test = False  # By default we are not in testing mode\n\n        testvar = str(os.environ.get(\"PARSL_TESTING\", 'None')).lower()\n        if testvar == 'true':\n            test = True\n\n        if not self.config.usage_tracking:\n            track = False\n\n        envvar = str(os.environ.get(\"PARSL_TRACKING\", True)).lower()\n        if envvar == \"false\":\n            track = False\n\n        return test, track\n\n    def construct_start_message(self):\n        \"\"\"Collect preliminary run info at the start of the DFK.\n\n        Returns :\n              - Message dict dumped as json string, ready for UDP\n        \"\"\"\n        uname = getpass.getuser().encode('latin1')\n        hashed_username = hashlib.sha256(uname).hexdigest()[0:10]\n        hname = socket.gethostname().encode('latin1')\n        hashed_hostname = hashlib.sha256(hname).hexdigest()[0:10]\n        message = {'uuid': self.uuid,\n                   'uname': hashed_username,\n                   'hname': hashed_hostname,\n                   'test': self.test_mode,\n                   'parsl_v': self.parsl_version,\n                   'python_v': self.python_version,\n                   'os': platform.system(),\n                   'os_v': platform.release(),\n                   'start': time.time()}\n\n        return json.dumps(message)\n\n    def construct_end_message(self):\n        \"\"\"Collect the final run information at the time of DFK cleanup.\n\n        Returns:\n             - Message dict dumped as json string, ready for UDP\n        \"\"\"\n        app_count = self.dfk.task_count\n\n        site_count = len([x for x in self.dfk.config.executors if x.managed])\n\n        app_fails = len([t for t in self.dfk.tasks if\n                         self.dfk.tasks[t]['status'] in FINAL_FAILURE_STATES])\n\n        message = {'uuid': self.uuid,\n                   'end': time.time(),\n                   't_apps': app_count,\n                   'sites': site_count,\n                   'c_time': None,\n                   'failed': app_fails,\n                   'test': self.test_mode,\n                   }\n\n        return json.dumps(message)\n\n    def send_UDP_message(self, message):\n        \"\"\"Send UDP message.\"\"\"\n        x = 0\n        if self.tracking_enabled:\n            try:\n                proc = udp_messenger(self.domain_name, self.UDP_IP, self.UDP_PORT, self.sock_timeout, message)\n                self.procs.append(proc)\n            except Exception as e:\n                logger.debug(\"Usage tracking failed: {}\".format(e))\n        else:\n            x = -1\n\n        return x\n\n    def send_message(self):\n        \"\"\"Send message over UDP.\n\n        If tracking is disables, the bytes_sent will always be set to -1\n\n        Returns:\n            (bytes_sent, time_taken)\n        \"\"\"\n        start = time.time()\n        message = None\n        if not self.initialized:\n            message = self.construct_start_message()\n            self.initialized = True\n        else:\n            message = self.construct_end_message()\n\n        self.send_UDP_message(message)\n        end = time.time()\n\n        return end - start\n\n    def __del__(self):\n        return self.close()\n\n    def close(self):\n        \"\"\"We terminate (SIGTERM) the processes added to the self.procs list \"\"\"\n        for proc in self.procs:\n            proc.terminate()\n\n\nif __name__ == '__main__':\n\n    from parsl import *\n\n    set_stream_logger()\n    workers = ThreadPoolExecutor(max_workers=4)\n    dfk = DataFlowKernel(executors=[workers])\n\n    dfk.cleanup()\n", "path": "parsl/dataflow/usage_tracking/usage.py"}]}