Consistently use salt.utils.json and salt.utils.msgpack.

This commit introduces salt.utils.msgpack modifies all places in the
code that either use json or msgpack to use salt.utils.json or
salt.utils.msgpack respectively.

While this change itself does not have any effect, it is important to
allow for centrally dealing with objects that cannot be directly
serialied via json or msgpack.

salt/cloud/clouds/ec2.py

@@ -92,7 +92,6 @@ import hashlib
 import binascii
 import datetime
 import base64
-import msgpack
 import re
 import decimal
 
@@ -102,6 +101,7 @@ import salt.utils.compat
 import salt.utils.files
 import salt.utils.hashutils
 import salt.utils.json
+import salt.utils.msgpack
 import salt.utils.stringutils
 import salt.utils.yaml
 from salt._compat import ElementTree as ET
@@ -5000,7 +5000,7 @@ def _parse_pricing(url, name):
         __opts__['cachedir'], 'ec2-pricing-{0}.p'.format(name)
     )
     with salt.utils.files.fopen(outfile, 'w') as fho:
-        msgpack.dump(regions, fho)
+        salt.utils.msgpack.dump(regions, fho)
 
     return True
 
@@ -5068,7 +5068,8 @@ def show_pricing(kwargs=None, call=None):
         update_pricing({'type': name}, 'function')
 
     with salt.utils.files.fopen(pricefile, 'r') as fhi:
-        ec2_price = salt.utils.stringutils.to_unicode(msgpack.load(fhi))
+        ec2_price = salt.utils.stringutils.to_unicode(
+            salt.utils.msgpack.load(fhi))
 
     region = get_location(profile)
     size = profile.get('size', None)

salt/cloud/clouds/gce.py

@@ -53,7 +53,6 @@ import sys
 import re
 import pprint
 import logging
-import msgpack
 from ast import literal_eval
 from salt.utils.versions import LooseVersion as _LooseVersion
 
@@ -91,6 +90,7 @@ from salt.ext import six
 import salt.utils.cloud
 import salt.utils.files
 import salt.utils.http
+import salt.utils.msgpack
 import salt.config as config
 from salt.cloud.libcloudfuncs import *  # pylint: disable=redefined-builtin,wildcard-import,unused-wildcard-import
 from salt.exceptions import (
@@ -2629,7 +2629,7 @@ def update_pricing(kwargs=None, call=None):
         __opts__['cachedir'], 'gce-pricing.p'
     )
     with salt.utils.files.fopen(outfile, 'w') as fho:
-        msgpack.dump(price_json['dict'], fho)
+        salt.utils.msgpack.dump(price_json['dict'], fho)
 
     return True
 
@@ -2668,7 +2668,7 @@ def show_pricing(kwargs=None, call=None):
         update_pricing()
 
     with salt.utils.files.fopen(pricefile, 'r') as fho:
-        sizes = msgpack.load(fho)
+        sizes = salt.utils.msgpack.load(fho)
 
     per_hour = float(sizes['gcp_price_list'][size][region])
 

salt/engines/stalekey.py

@@ -28,11 +28,11 @@ import salt.config
 import salt.key
 import salt.utils.files
 import salt.utils.minions
+import salt.utils.msgpack
 import salt.wheel
 
 # Import 3rd-party libs
 from salt.ext import six
-import msgpack
 
 log = logging.getLogger(__name__)
 
@@ -60,7 +60,7 @@ def start(interval=3600, expire=604800):
         if os.path.exists(presence_file):
             try:
                 with salt.utils.files.fopen(presence_file, 'r') as f:
-                    minions = msgpack.load(f)
+                    minions = salt.utils.msgpack.load(f)
             except IOError as e:
                 log.error('Could not open presence file %s: %s', presence_file, e)
                 time.sleep(interval)
@@ -95,7 +95,7 @@ def start(interval=3600, expire=604800):
 
         try:
             with salt.utils.files.fopen(presence_file, 'w') as f:
-                msgpack.dump(minions, f)
+                salt.utils.msgpack.dump(minions, f)
         except IOError as e:
             log.error('Could not write to presence file %s: %s', presence_file, e)
         time.sleep(interval)

salt/log/handlers/fluent_mod.py

@@ -96,14 +96,17 @@ log = logging.getLogger(__name__)
 try:
     # Attempt to import msgpack
     import msgpack
+    import salt.utils.msgpack
     # There is a serialization issue on ARM and potentially other platforms
     # for some msgpack bindings, check for it
     if msgpack.loads(msgpack.dumps([1, 2, 3]), use_list=True) is None:
         raise ImportError
+    import salt.utils.msgpack
 except ImportError:
     # Fall back to msgpack_pure
     try:
         import msgpack_pure as msgpack
+        import salt.utils.msgpack
     except ImportError:
         # TODO: Come up with a sane way to get a configured logfile
         #       and write to the logfile when this error is hit also
@@ -455,7 +458,7 @@ class FluentSender(object):
         packet = (tag, timestamp, data)
         if self.verbose:
             print(packet)
-        return msgpack.packb(packet)
+        return salt.utils.msgpack.packb(packet, _msgpack_module=msgpack)
 
     def _send(self, bytes_):
         self.lock.acquire()

salt/modules/saltcheck.py

@@ -144,7 +144,7 @@ from __future__ import absolute_import, unicode_literals, print_function
 import logging
 import os
 import time
-from json import loads, dumps
+from salt.utils.json import loads, dumps
 
 # Import Salt libs
 import salt.utils.files

salt/modules/state.py

@@ -33,6 +33,7 @@ import salt.utils.functools
 import salt.utils.hashutils
 import salt.utils.jid
 import salt.utils.json
+import salt.utils.msgpack
 import salt.utils.platform
 import salt.utils.state
 import salt.utils.stringutils
@@ -45,7 +46,6 @@ from salt.utils.odict import OrderedDict
 
 # Import 3rd-party libs
 from salt.ext import six
-import msgpack
 
 __proxyenabled__ = ['*']
 
@@ -185,7 +185,7 @@ def _get_pause(jid, state_id=None):
             data[state_id] = {}
     if os.path.exists(pause_path):
         with salt.utils.files.fopen(pause_path, 'rb') as fp_:
-            data = msgpack.loads(fp_.read())
+            data = salt.utils.msgpack.loads(fp_.read())
     return data, pause_path
 
 
@@ -256,7 +256,7 @@ def soft_kill(jid, state_id=None):
     data, pause_path = _get_pause(jid, state_id)
     data[state_id]['kill'] = True
     with salt.utils.files.fopen(pause_path, 'wb') as fp_:
-        fp_.write(msgpack.dumps(data))
+        fp_.write(salt.utils.msgpack.dumps(data))
 
 
 def pause(jid, state_id=None, duration=None):
@@ -291,7 +291,7 @@ def pause(jid, state_id=None, duration=None):
     if duration:
         data[state_id]['duration'] = int(duration)
     with salt.utils.files.fopen(pause_path, 'wb') as fp_:
-        fp_.write(msgpack.dumps(data))
+        fp_.write(salt.utils.msgpack.dumps(data))
 
 
 def resume(jid, state_id=None):
@@ -325,7 +325,7 @@ def resume(jid, state_id=None):
     if state_id == '__all__':
         data = {}
     with salt.utils.files.fopen(pause_path, 'wb') as fp_:
-        fp_.write(msgpack.dumps(data))
+        fp_.write(salt.utils.msgpack.dumps(data))
 
 
 def orchestrate(mods,

salt/modules/winrepo.py

@@ -32,11 +32,8 @@ from salt.runners.winrepo import (
     GLOBAL_ONLY
 )
 from salt.ext import six
-try:
-    import msgpack
-except ImportError:
-    import msgpack_pure as msgpack  # pylint: disable=import-error
 import salt.utils.gitfs
+import salt.utils.msgpack
 # pylint: enable=unused-import
 
 log = logging.getLogger(__name__)

salt/payload.py

@@ -59,6 +59,10 @@ except ImportError:
         #sys.exit(salt.defaults.exitcodes.EX_GENERIC)
 
 
+if HAS_MSGPACK:
+    import salt.utils.msgpack
+
+
 if HAS_MSGPACK and not hasattr(msgpack, 'exceptions'):
     class PackValueError(Exception):
         '''
@@ -79,14 +83,15 @@ def package(payload):
     This method for now just wraps msgpack.dumps, but it is here so that
     we can make the serialization a custom option in the future with ease.
     '''
-    return msgpack.dumps(payload)
+    return salt.utils.msgpack.dumps(payload, _msgpack_module=msgpack)
 
 
 def unpackage(package_):
     '''
     Unpackages a payload
     '''
-    return msgpack.loads(package_, use_list=True)
+    return salt.utils.msgpack.loads(package_, use_list=True,
+                                    _msgpack_module=msgpack)
 
 
 def format_payload(enc, **kwargs):
@@ -155,14 +160,19 @@ class Serial(object):
                 else:
                     loads_kwargs['encoding'] = encoding
                 try:
-                    ret = msgpack.loads(msg, **loads_kwargs)
+                    ret = salt.utils.msgpack.loads(msg, use_list=True,
+                                                   ext_hook=ext_type_decoder,
+                                                   encoding=encoding,
+                                                   _msgpack_module=msgpack)
                 except UnicodeDecodeError:
                     # msg contains binary data
                     loads_kwargs.pop('raw', None)
                     loads_kwargs.pop('encoding', None)
                     ret = msgpack.loads(msg, **loads_kwargs)
             else:
-                ret = msgpack.loads(msg, **loads_kwargs)
+                ret = salt.utils.msgpack.loads(msg, use_list=True,
+                                               ext_hook=ext_type_decoder,
+                                               _msgpack_module=msgpack)
             if six.PY3 and encoding is None and not raw:
                 ret = salt.transport.frame.decode_embedded_strs(ret)
         except Exception as exc:
@@ -237,9 +247,12 @@ class Serial(object):
                 # Due to this, if we don't need it, don't pass it at all so
                 # that under Python 2 we can still work with older versions
                 # of msgpack.
-                return msgpack.dumps(msg, default=ext_type_encoder, use_bin_type=use_bin_type)
+                return salt.utils.msgpack.dumps(msg, default=ext_type_encoder,
+                                                use_bin_type=use_bin_type,
+                                                _msgpack_module=msgpack)
             else:
-                return msgpack.dumps(msg, default=ext_type_encoder)
+                return salt.utils.msgpack.dumps(msg, default=ext_type_encoder,
+                                                _msgpack_module=msgpack)
         except (OverflowError, msgpack.exceptions.PackValueError):
             # msgpack<=0.4.6 don't call ext encoder on very long integers raising the error instead.
             # Convert any very long longs to strings and call dumps again.
@@ -268,9 +281,12 @@ class Serial(object):
 
             msg = verylong_encoder(msg, set())
             if msgpack.version >= (0, 4, 0):
-                return msgpack.dumps(msg, default=ext_type_encoder, use_bin_type=use_bin_type)
+                return salt.utils.msgpack.dumps(msg, default=ext_type_encoder,
+                                                use_bin_type=use_bin_type,
+                                                _msgpack_module=msgpack)
             else:
-                return msgpack.dumps(msg, default=ext_type_encoder)
+                return salt.utils.msgpack.dumps(msg, default=ext_type_encoder,
+                                                _msgpack_module=msgpack)
 
     def dump(self, msg, fn_):
         '''

salt/renderers/msgpack.py

@@ -1,10 +1,8 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import, print_function, unicode_literals
 
-# Import third party libs
-import msgpack
-
 # Import salt libs
+import salt.utils.msgpack
 from salt.ext import six
 
 
@@ -28,4 +26,4 @@ def render(msgpack_data, saltenv='base', sls='', **kws):
         msgpack_data = msgpack_data[(msgpack_data.find('\n') + 1):]
     if not msgpack_data.strip():
         return {}
-    return msgpack.loads(msgpack_data)
+    return salt.utils.msgpack.loads(msgpack_data)

salt/returners/local_cache.py

@@ -20,11 +20,11 @@ import salt.utils.atomicfile
 import salt.utils.files
 import salt.utils.jid
 import salt.utils.minions
+import salt.utils.msgpack
 import salt.utils.stringutils
 import salt.exceptions
 
 # Import 3rd-party libs
-import msgpack
 from salt.ext import six
 from salt.ext.six.moves import range  # pylint: disable=import-error,redefined-builtin
 
@@ -520,7 +520,7 @@ def save_reg(data):
             raise
     try:
         with salt.utils.files.fopen(regfile, 'a') as fh_:
-            msgpack.dump(data, fh_)
+            salt.utils.msgpack.dump(data, fh_)
     except Exception:
         log.error('Could not write to msgpack file %s', __opts__['outdir'])
         raise
@@ -534,7 +534,7 @@ def load_reg():
     regfile = os.path.join(reg_dir, 'register')
     try:
         with salt.utils.files.fopen(regfile, 'r') as fh_:
-            return msgpack.load(fh_)
+            return salt.utils.msgpack.load(fh_)
     except Exception:
         log.error('Could not write to msgpack file %s', __opts__['outdir'])
         raise

salt/runners/winrepo.py

@@ -12,15 +12,12 @@ import os
 
 # Import third party libs
 from salt.ext import six
-try:
-    import msgpack
-except ImportError:
-    import msgpack_pure as msgpack  # pylint: disable=import-error
 
 # Import salt libs
 from salt.exceptions import CommandExecutionError, SaltRenderError
 import salt.utils.files
 import salt.utils.gitfs
+import salt.utils.msgpack
 import salt.utils.path
 import logging
 import salt.minion
@@ -124,7 +121,7 @@ def genrepo(opts=None, fire_event=True):
                     ret.setdefault('name_map', {}).update(revmap)
     with salt.utils.files.fopen(
             os.path.join(winrepo_dir, winrepo_cachefile), 'w+b') as repo:
-        repo.write(msgpack.dumps(ret))
+        repo.write(salt.utils.msgpack.dumps(ret))
     return ret
 
 

salt/sdb/sqlite3.py

@@ -54,11 +54,9 @@ except ImportError:
     HAS_SQLITE3 = False
 
 # Import salt libs
+import salt.utils.msgpack
 from salt.ext import six
 
-# Import third party libs
-import msgpack
-
 
 DEFAULT_TABLE = 'sdb'
 
@@ -126,9 +124,9 @@ def set_(key, value, profile=None):
         return False
     conn, cur, table = _connect(profile)
     if six.PY2:
-        value = buffer(msgpack.packb(value))
+        value = buffer(salt.utils.msgpack.packb(value))
     else:
-        value = memoryview(msgpack.packb(value))
+        value = memoryview(salt.utils.msgpack.packb(value))
     q = profile.get('set_query', ('INSERT OR REPLACE INTO {0} VALUES '
                                   '(:key, :value)').format(table))
     conn.execute(q, {'key': key, 'value': value})
@@ -149,4 +147,4 @@ def get(key, profile=None):
     res = res.fetchone()
     if not res:
         return None
-    return msgpack.unpackb(res[0])
+    return salt.utils.msgpack.unpackb(res[0])

salt/serializers/msgpack.py

@@ -24,6 +24,7 @@ log = logging.getLogger(__name__)
 try:
     # Attempt to import msgpack
     import msgpack
+    import salt.utils.msgpack
     # There is a serialization issue on ARM and potentially other platforms
     # for some msgpack bindings, check for it
     if msgpack.loads(msgpack.dumps([1, 2, 3]), use_list=True) is None:
@@ -33,6 +34,7 @@ except ImportError:
     # Fall back to msgpack_pure
     try:
         import msgpack_pure as msgpack  # pylint: disable=import-error
+        import salt.utils.msgpack
     except ImportError:
         # TODO: Come up with a sane way to get a configured logfile
         #       and write to the logfile when this error is hit also
@@ -60,7 +62,8 @@ elif msgpack.version >= (0, 2, 0):
 
     def _serialize(obj, **options):
         try:
-            return msgpack.dumps(obj, **options)
+            return salt.utils.msgpack.dumps(obj, _msgpack_module=msgpack,
+                                            **options)
         except Exception as error:
             raise SerializationError(error)
 
@@ -68,7 +71,9 @@ elif msgpack.version >= (0, 2, 0):
         try:
             options.setdefault('use_list', True)
             options.setdefault('encoding', 'utf-8')
-            return msgpack.loads(stream_or_string, **options)
+            return salt.utils.msgpack.loads(stream_or_string,
+                                            _msgpack_module=msgpack,
+                                            **options)
         except Exception as error:
             raise DeserializationError(error)
 
@@ -95,14 +100,16 @@ else:  # msgpack.version < 0.2.0
     def _serialize(obj, **options):
         try:
             obj = _encoder(obj)
-            return msgpack.dumps(obj, **options)
+            return salt.utils.msgpack.dumps(obj, _msgpack_module=msgpack,
+                                            **options)
         except Exception as error:
             raise SerializationError(error)
 
     def _deserialize(stream_or_string, **options):
         options.setdefault('use_list', True)
         try:
-            obj = msgpack.loads(stream_or_string)
+            obj = salt.utils.msgpack.loads(stream_or_string,
+                                           _msgpack_module=msgpack)
             return _decoder(obj)
         except Exception as error:
             raise DeserializationError(error)

salt/state.py

@@ -40,6 +40,7 @@ import salt.utils.event
 import salt.utils.files
 import salt.utils.hashutils
 import salt.utils.immutabletypes as immutabletypes
+import salt.utils.msgpack as msgpack
 import salt.utils.platform
 import salt.utils.process
 import salt.utils.url
@@ -56,7 +57,6 @@ from salt.utils.odict import OrderedDict, DefaultOrderedDict
 import salt.utils.yamlloader as yamlloader
 
 # Import third party libs
-import msgpack
 # pylint: disable=import-error,no-name-in-module,redefined-builtin
 from salt.ext import six
 from salt.ext.six.moves import map, range, reload_module

salt/states/netsnmp.py

@@ -23,9 +23,8 @@ from __future__ import absolute_import, print_function, unicode_literals
 import logging
 log = logging.getLogger(__name__)
 
-from json import loads, dumps
-
 # salt lib
+from salt.utils.json import loads, dumps
 from salt.ext import six
 # import NAPALM utils
 import salt.utils.napalm

salt/states/netusers.py

@@ -25,9 +25,9 @@ log = logging.getLogger(__name__)
 
 # Python std lib
 from copy import deepcopy
-from json import loads, dumps
 
 # salt lib
+from salt.utils.json import loads, dumps
 from salt.ext import six
 # import NAPALM utils
 import salt.utils.napalm

salt/states/pkg.py

@@ -135,10 +135,6 @@ if salt.utils.platform.is_windows():
     # The following imports are used by the namespaced win_pkg funcs
     # and need to be included in their globals.
     # pylint: disable=import-error,unused-import
-    try:
-        import msgpack
-    except ImportError:
-        import msgpack_pure as msgpack
     from salt.utils.versions import LooseVersion
     # pylint: enable=import-error,unused-import
 # pylint: enable=invalid-name

salt/states/probes.py

@@ -25,9 +25,9 @@ import logging
 log = logging.getLogger(__name__)
 
 from copy import deepcopy
-from json import loads, dumps
 
 # salt modules
+from salt.utils.json import loads, dumps
 from salt.ext import six
 # import NAPALM utils
 import salt.utils.napalm

salt/states/zabbix_host.py

@@ -7,7 +7,7 @@ Management of Zabbix hosts.
 
 '''
 from __future__ import absolute_import, print_function, unicode_literals
-from json import loads, dumps
+from salt.utils.json import loads, dumps
 from copy import deepcopy
 from salt.ext import six
 

salt/states/zabbix_user.py

@@ -9,7 +9,7 @@ Management of Zabbix users.
 
 # Import Python libs
 from __future__ import absolute_import, print_function, unicode_literals
-from json import loads, dumps
+from salt.utils.json import loads, dumps
 from copy import deepcopy
 
 # Import Salt libs

salt/transport/frame.py

@@ -4,7 +4,7 @@ Helper functions for transport components to handle message framing
 '''
 # Import python libs
 from __future__ import absolute_import, print_function, unicode_literals
-import msgpack
+import salt.utils.msgpack
 from salt.ext import six
 
 
@@ -18,7 +18,7 @@ def frame_msg(body, header=None, raw_body=False):  # pylint: disable=unused-argu
 
     framed_msg['head'] = header
     framed_msg['body'] = body
-    return msgpack.dumps(framed_msg)
+    return salt.utils.msgpack.dumps(framed_msg)
 
 
 def frame_msg_ipc(body, header=None, raw_body=False):  # pylint: disable=unused-argument
@@ -35,9 +35,9 @@ def frame_msg_ipc(body, header=None, raw_body=False):  # pylint: disable=unused-
     framed_msg['head'] = header
     framed_msg['body'] = body
     if six.PY2:
-        return msgpack.dumps(framed_msg)
+        return salt.utils.msgpack.dumps(framed_msg)
     else:
-        return msgpack.dumps(framed_msg, use_bin_type=True)
+        return salt.utils.msgpack.dumps(framed_msg, use_bin_type=True)
 
 
 def _decode_embedded_list(src):

salt/transport/ipc.py

@@ -11,9 +11,6 @@ import logging
 import socket
 import time
 
-# Import 3rd-party libs
-import msgpack
-
 # Import Tornado libs
 import tornado
 import tornado.gen
@@ -25,6 +22,7 @@ from tornado.iostream import IOStream, StreamClosedError
 # Import Salt libs
 import salt.transport.client
 import salt.transport.frame
+import salt.utils.msgpack as msgpack
 from salt.ext import six
 
 log = logging.getLogger(__name__)

salt/transport/tcp.py

@@ -23,6 +23,7 @@ import salt.crypt
 import salt.utils.asynchronous
 import salt.utils.event
 import salt.utils.files
+import salt.utils.msgpack as msgpack
 import salt.utils.platform
 import salt.utils.process
 import salt.utils.verify

salt/utils/cache.py

@@ -9,8 +9,7 @@ import re
 import time
 import logging
 try:
-    import msgpack
-    HAS_MSGPACK = True
+    import salt.utils.msgpack as msgpack
 except ImportError:
     HAS_MSGPACK = False
 

salt/utils/cloud.py

@@ -22,6 +22,11 @@ import subprocess
 import sys
 import tempfile
 import time
+import subprocess
+import multiprocessing
+import logging
+import pipes
+import msgpack
 import traceback
 import uuid
 
@@ -64,6 +69,7 @@ import salt.utils.data
 import salt.utils.event
 import salt.utils.files
 import salt.utils.path
+import salt.utils.msgpack
 import salt.utils.platform
 import salt.utils.stringutils
 import salt.utils.versions
@@ -2629,7 +2635,9 @@ def cachedir_index_add(minion_id, profile, driver, provider, base=None):
     if os.path.exists(index_file):
         mode = 'rb' if six.PY3 else 'r'
         with salt.utils.files.fopen(index_file, mode) as fh_:
-            index = salt.utils.data.decode(msgpack.load(fh_, encoding=MSGPACK_ENCODING))
+            index = salt.utils.data.decode(
+                salt.utils.msgpack.msgpack.load(
+                    fh_, encoding=MSGPACK_ENCODING))
     else:
         index = {}
 
@@ -2646,7 +2654,7 @@ def cachedir_index_add(minion_id, profile, driver, provider, base=None):
 
     mode = 'wb' if six.PY3 else 'w'
     with salt.utils.files.fopen(index_file, mode) as fh_:
-        msgpack.dump(index, fh_, encoding=MSGPACK_ENCODING)
+        salt.utils.msgpack.dump(index, fh_, encoding=MSGPACK_ENCODING)
 
     unlock_file(index_file)
 
@@ -2663,7 +2671,8 @@ def cachedir_index_del(minion_id, base=None):
     if os.path.exists(index_file):
         mode = 'rb' if six.PY3 else 'r'
         with salt.utils.files.fopen(index_file, mode) as fh_:
-            index = salt.utils.data.decode(msgpack.load(fh_, encoding=MSGPACK_ENCODING))
+            index = salt.utils.data.decode(
+                salt.utils.msgpack.load(fh_, encoding=MSGPACK_ENCODING))
     else:
         return
 
@@ -2672,7 +2681,7 @@ def cachedir_index_del(minion_id, base=None):
 
     mode = 'wb' if six.PY3 else 'w'
     with salt.utils.files.fopen(index_file, mode) as fh_:
-        msgpack.dump(index, fh_, encoding=MSGPACK_ENCODING)
+        salt.utils.msgpack.dump(index, fh_, encoding=MSGPACK_ENCODING)
 
     unlock_file(index_file)
 
@@ -2730,7 +2739,7 @@ def request_minion_cachedir(
     path = os.path.join(base, 'requested', fname)
     mode = 'wb' if six.PY3 else 'w'
     with salt.utils.files.fopen(path, mode) as fh_:
-        msgpack.dump(data, fh_, encoding=MSGPACK_ENCODING)
+        salt.utils.msgpack.dump(data, fh_, encoding=MSGPACK_ENCODING)
 
 
 def change_minion_cachedir(
@@ -2762,12 +2771,13 @@ def change_minion_cachedir(
     path = os.path.join(base, cachedir, fname)
 
     with salt.utils.files.fopen(path, 'r') as fh_:
-        cache_data = salt.utils.data.decode(msgpack.load(fh_, encoding=MSGPACK_ENCODING))
+        cache_data = salt.utils.data.decode(
+            salt.utils.msgpack.load(fh_, encoding=MSGPACK_ENCODING))
 
     cache_data.update(data)
 
     with salt.utils.files.fopen(path, 'w') as fh_:
-        msgpack.dump(cache_data, fh_, encoding=MSGPACK_ENCODING)
+        salt.utils.msgpack.dump(cache_data, fh_, encoding=MSGPACK_ENCODING)
 
 
 def activate_minion_cachedir(minion_id, base=None):
@@ -2841,7 +2851,8 @@ def list_cache_nodes_full(opts=None, provider=None, base=None):
                 minion_id = fname[:-2]  # strip '.p' from end of msgpack filename
                 mode = 'rb' if six.PY3 else 'r'
                 with salt.utils.files.fopen(fpath, mode) as fh_:
-                    minions[driver][prov][minion_id] = salt.utils.data.decode(msgpack.load(fh_, encoding=MSGPACK_ENCODING))
+                    minions[driver][prov][minion_id] = salt.utils.data.decode(
+                        salt.utils.msgpack.load(fh_, encoding=MSGPACK_ENCODING))
 
     return minions
 
@@ -3002,7 +3013,7 @@ def cache_node_list(nodes, provider, opts):
         path = os.path.join(prov_dir, '{0}.p'.format(node))
         mode = 'wb' if six.PY3 else 'w'
         with salt.utils.files.fopen(path, mode) as fh_:
-            msgpack.dump(nodes[node], fh_, encoding=MSGPACK_ENCODING)
+            salt.utils.msgpack.dump(nodes[node], fh_, encoding=MSGPACK_ENCODING)
 
 
 def cache_node(node, provider, opts):
@@ -3028,7 +3039,7 @@ def cache_node(node, provider, opts):
     path = os.path.join(prov_dir, '{0}.p'.format(node['name']))
     mode = 'wb' if six.PY3 else 'w'
     with salt.utils.files.fopen(path, mode) as fh_:
-        msgpack.dump(node, fh_, encoding=MSGPACK_ENCODING)
+        salt.utils.msgpack.dump(node, fh_, encoding=MSGPACK_ENCODING)
 
 
 def missing_node_cache(prov_dir, node_list, provider, opts):
@@ -3103,7 +3114,8 @@ def diff_node_cache(prov_dir, node, new_data, opts):
 
     with salt.utils.files.fopen(path, 'r') as fh_:
         try:
-            cache_data = salt.utils.data.decode(msgpack.load(fh_, encoding=MSGPACK_ENCODING))
+            cache_data = salt.utils.data.decode(
+                salt.utils.msgpack.load(fh_, encoding=MSGPACK_ENCODING))
         except ValueError:
             log.warning('Cache for %s was corrupt: Deleting', node)
             cache_data = {}

salt/utils/http.py

@@ -84,7 +84,7 @@ except ImportError:
     HAS_REQUESTS = False
 
 try:
-    import msgpack
+    import salt.utils.msgpack
     HAS_MSGPACK = True
 except ImportError:
     HAS_MSGPACK = False
@@ -277,12 +277,12 @@ def query(url,
         # contain expirations, they can't be stored in a proper cookie jar.
         if os.path.isfile(session_cookie_jar):
             with salt.utils.files.fopen(session_cookie_jar, 'rb') as fh_:
-                session_cookies = msgpack.load(fh_)
+                session_cookies = salt.utils.msgpack.load(fh_)
             if isinstance(session_cookies, dict):
                 header_dict.update(session_cookies)
         else:
             with salt.utils.files.fopen(session_cookie_jar, 'wb') as fh_:
-                msgpack.dump('', fh_)
+                salt.utils.msgpack.dump('', fh_)
 
     for header in header_list:
         comps = header.split(':')
@@ -656,9 +656,9 @@ def query(url,
             with salt.utils.files.fopen(session_cookie_jar, 'wb') as fh_:
                 session_cookies = result_headers.get('set-cookie', None)
                 if session_cookies is not None:
-                    msgpack.dump({'Cookie': session_cookies}, fh_)
+                    salt.utils.msgpack.dump({'Cookie': session_cookies}, fh_)
                 else:
-                    msgpack.dump('', fh_)
+                    salt.utils.msgpack.dump('', fh_)
 
     if status is True:
         ret['status'] = result_status_code

salt/utils/msgpack.py

@@ -0,0 +1,80 @@
+# -*- coding: utf-8 -*-
+'''
+Functions to work with MessagePack
+'''
+
+from __future__ import absolute_import
+
+# Import Python libs
+try:
+    # Attempt to import msgpack
+    import msgpack
+except ImportError:
+    # Fall back to msgpack_pure
+    import msgpack_pure as msgpack  # pylint: disable=import-error
+
+
+def pack(o, stream, **kwargs):
+    '''
+    .. versionadded:: 2018.3.4
+
+    Wraps msgpack.pack and ensures that the passed object is unwrapped if it is
+    a proxy.
+
+    By default, this function uses the msgpack module and falls back to
+    msgpack_pure, if the msgpack is not available. You can pass an alternate
+    msgpack module using the _msgpack_module argument.
+    '''
+    msgpack_module = kwargs.pop('_msgpack_module', msgpack)
+    return msgpack_module.pack(o, stream, **kwargs)
+
+
+def packb(o, **kwargs):
+    '''
+    .. versionadded:: 2018.3.4
+
+    Wraps msgpack.packb and ensures that the passed object is unwrapped if it
+    is a proxy.
+
+    By default, this function uses the msgpack module and falls back to
+    msgpack_pure, if the msgpack is not available. You can pass an alternate
+    msgpack module using the _msgpack_module argument.
+    '''
+    msgpack_module = kwargs.pop('_msgpack_module', msgpack)
+    return msgpack_module.packb(o, **kwargs)
+
+
+def unpack(stream, **kwargs):
+    '''
+    .. versionadded:: 2018.3.4
+
+    Wraps msgpack.unpack.
+
+    By default, this function uses the msgpack module and falls back to
+    msgpack_pure, if the msgpack is not available. You can pass an alternate
+    msgpack module using the _msgpack_module argument.
+    '''
+    msgpack_module = kwargs.pop('_msgpack_module', msgpack)
+    return msgpack_module.unpack(stream, **kwargs)
+
+
+def unpackb(packed, **kwargs):
+    '''
+    .. versionadded:: 2018.3.4
+
+    Wraps msgpack.unpack.
+
+    By default, this function uses the msgpack module and falls back to
+    msgpack_pure, if the msgpack is not available. You can pass an alternate
+    msgpack module using the _msgpack_module argument.
+    '''
+    msgpack_module = kwargs.pop('_msgpack_module', msgpack)
+    return msgpack_module.unpackb(packed, **kwargs)
+
+
+# alias for compatibility to simplejson/marshal/pickle.
+load = unpack
+loads = unpackb
+
+dump = pack
+dumps = packb

tests/integration/__init__.py

@@ -52,6 +52,7 @@ import salt.output
 import salt.version
 import salt.utils.color
 import salt.utils.files
+import salt.utils.msgpack as msgpack
 import salt.utils.path
 import salt.utils.platform
 import salt.utils.process
@@ -63,7 +64,6 @@ from salt.utils.immutabletypes import freeze
 from salt.exceptions import SaltClientError
 
 # Import 3rd-party libs
-import msgpack
 from salt.ext import six
 
 try:

tests/integration/files/log_handlers/runtests_log_handler.py

@@ -19,10 +19,8 @@ import logging
 import threading
 from multiprocessing import Queue
 
-# Import 3rd-party libs
-import msgpack
-
 # Import Salt libs
+import salt.utils.msgpack
 from salt.ext import six
 from salt.utils.platform import is_darwin
 import salt.log.setup
@@ -95,7 +93,8 @@ def process_queue(port, queue):
                 break
             # Just log everything, filtering will happen on the main process
             # logging handlers
-            sock.sendall(msgpack.dumps(record.__dict__, encoding='utf-8'))
+            sock.sendall(salt.utils.msgpack.dumps(record.__dict__,
+                                                  encoding='utf-8'))
         except (IOError, EOFError, KeyboardInterrupt, SystemExit):
             try:
                 sock.shutdown(socket.SHUT_RDWR)

tests/packdump.py

@@ -9,8 +9,8 @@ import os
 import sys
 import pprint
 
-# Import third party libs
-import msgpack
+# Import Salt libs
+import salt.utils.msgpack
 
 
 def dump(path):
@@ -21,7 +21,7 @@ def dump(path):
         print('Not a file')
         return
     with open(path, 'rb') as fp_:
-        data = msgpack.loads(fp_.read())
+        data = salt.utils.msgpack.loads(fp_.read())
         pprint.pprint(data)