This function needs to correspond to this:
        https://github.com/saltstack/salt/blob/develop/salt/returners/redis_return.py#L88
330731
def _write(self, ret):
        
        self.redis.set('{0}:{1}'.format(ret['id'], ret['jid']), json.dumps(ret))
        self.redis.lpush('{0}:{1}'.format(ret['id'], ret['fun']), ret['jid'])
        self.redis.sadd('minions', ret['id'])
        self.redis.sadd('jids', ret['jid'])
294038
def kdf_salt(self):
        

        encryption_algo = self['algorithm'].native

        if encryption_algo == 'pbes2':
            salt = self['parameters']['key_derivation_func']['parameters']['salt']

            if salt.name == 'other_source':
                raise ValueError(unwrap(
                    '''
                    Can not determine key derivation salt - the
                    reserved-for-future-use other source salt choice was
                    specified in the PBKDF2 params structure
                    '''
                ))

            return salt.native

        if encryption_algo.find('.') == -1:
            if encryption_algo.find('_') != -1:
                return self['parameters']['salt'].native

            raise ValueError(unwrap(
                '''
                Encryption algorithm "%s" does not have a registered key
                derivation function
                ''',
                encryption_algo
            ))

        raise ValueError(unwrap(
            '''
            Unrecognized encryption algorithm "%s", can not determine key
            derivation salt
            ''',
            encryption_algo
        ))
1319580
def ext_pillar(minion_id, pillar, function, **kwargs):
    '''
    
    '''
    if function.startswith('_') or function not in globals():
        return {}
    # Call specified function to pull redis data
    return globals()[function](minion_id, pillar, **kwargs)
470358
def render(data, saltenv='base', sls='', argline='', **kwargs):  # pylint: disable=unused-argument
    '''
    
    '''
    translate_newlines = kwargs.get('translate_newlines', False)
    return _decrypt_object(data, translate_newlines=translate_newlines)
582269
def _get_serv(ret=None, commit=False):
    '''
    
    '''
    _options = _get_options(ret)
    try:
        conn = psycopg2.connect(host=_options.get('host'),
                                user=_options.get('user'),
                                password=_options.get('passwd'),
                                database=_options.get('db'),
                                port=_options.get('port'))

    except psycopg2.OperationalError as exc:
        raise salt.exceptions.SaltMasterError('postgres returner could not connect to database: {exc}'.format(exc=exc))

    cursor = conn.cursor()

    try:
        yield cursor
    except psycopg2.DatabaseError as err:
        error = err.args
        sys.stderr.write(six.text_type(error))
        cursor.execute("ROLLBACK")
        raise err
    else:
        if commit:
            cursor.execute("COMMIT")
        else:
            cursor.execute("ROLLBACK")
    finally:
        conn.close()
1164937
def block_pipeline_command(func):
    

    def inner(*args, **kwargs):
        raise RedisClusterException(
            "ERROR: Calling pipelined function {0} is blocked when running redis in cluster mode...".format(
                func.__name__))

    return inner
1351611
def reap_fileserver_cache_dir(cache_base, find_func):
    '''
    
    '''
    for saltenv in os.listdir(cache_base):
        env_base = os.path.join(cache_base, saltenv)
        for root, dirs, files in salt.utils.path.os_walk(env_base):
            # if we have an empty directory, lets cleanup
            # This will only remove the directory on the second time
            # "_reap_cache" is called (which is intentional)
            if not dirs and not files:
                # only remove if empty directory is older than 60s
                if time.time() - os.path.getctime(root) > 60:
                    os.rmdir(root)
                continue
            # if not, lets check the files in the directory
            for file_ in files:
                file_path = os.path.join(root, file_)
                file_rel_path = os.path.relpath(file_path, env_base)
                try:
                    filename, _, hash_type = file_rel_path.rsplit('.', 2)
                except ValueError:
                    log.warning(
                        'Found invalid hash file [%s] when attempting to reap '
                        'cache directory', file_
                    )
                    continue
                # do we have the file?
                ret = find_func(filename, saltenv=saltenv)
                # if we don't actually have the file, lets clean up the cache
                # object
                if ret['path'] == '':
                    os.unlink(file_path)
1326881
def _get_serv(ret=None):
    '''
    
    '''
    _options = _get_options(ret)
    global REDIS_POOL
    if REDIS_POOL:
        return REDIS_POOL
    elif _options.get('cluster_mode'):
        REDIS_POOL = StrictRedisCluster(startup_nodes=_options.get('startup_nodes'),
                                        skip_full_coverage_check=_options.get('skip_full_coverage_check'),
                                        decode_responses=True)
    else:
        REDIS_POOL = redis.StrictRedis(host=_options.get('host'),
                                       port=_options.get('port'),
                                       unix_socket_path=_options.get('unix_socket_path', None),
                                       db=_options.get('db'),
                                       decode_responses=True,
                                       password=_options.get('password'))
    return REDIS_POOL
576949
def hset(self, name, key, value):
        
        with self.pipe as pipe:
            value = self._value_encode(key, value)
            key = self.memberparse.encode(key)
            return pipe.hset(self.redis_key(name), key, value)
996785
def _reorder(string, salt):
    """"""
    len_salt = len(salt)

    if len_salt != 0:
        string = list(string)
        index, integer_sum = 0, 0
        for i in range(len(string) - 1, 0, -1):
            integer = ord(salt[index])
            integer_sum += integer
            j = (integer + index + integer_sum) % i
            string[i], string[j] = string[j], string[i]
            index = (index + 1) % len_salt
        string = ''.join(string)

    return string
735373
def A(host, nameserver=None):
    '''
    
    '''
    if _has_dig():
        return __salt__['dig.A'](host, nameserver)
    elif nameserver is None:
        # fall back to the socket interface, if we don't care who resolves
        try:
            addresses = [sock[4][0] for sock in socket.getaddrinfo(host, None, socket.AF_INET, 0, socket.SOCK_RAW)]
            return addresses
        except socket.gaierror:
            return 'Unable to resolve {0}'.format(host)

    return 'This function requires dig, which is not currently available'
301427
def returner(ret):
    '''
    
    '''
    query = '''INSERT INTO {keyspace}.salt_returns (
                 jid, minion_id, fun, alter_time, full_ret, return, success
               ) VALUES (?, ?, ?, ?, ?, ?, ?)'''.format(keyspace=_get_keyspace())

    statement_arguments = ['{0}'.format(ret['jid']),
                           '{0}'.format(ret['id']),
                           '{0}'.format(ret['fun']),
                           int(time.time() * 1000),
                           salt.utils.json.dumps(ret).replace("'", "''"),
                           salt.utils.json.dumps(ret['return']).replace("'", "''"),
                           ret.get('success', False)]

    # cassandra_cql.cql_query may raise a CommandExecutionError
    try:
        __salt__['cassandra_cql.cql_query_with_prepare'](query,
                                                         'returner_return',
                                                         tuple(statement_arguments),
                                                         asynchronous=True)
    except CommandExecutionError:
        log.critical('Could not insert into salt_returns with Cassandra returner.')
        raise
    except Exception as e:
        log.critical('Unexpected error while inserting into salt_returns: %s', e)
        raise

    # Store the last function called by the minion
    # The data in salt.minions will be used by get_fun and get_minions
    query = '''INSERT INTO {keyspace}.minions (
                 minion_id, last_fun
               ) VALUES (?, ?)'''.format(keyspace=_get_keyspace())

    statement_arguments = ['{0}'.format(ret['id']), '{0}'.format(ret['fun'])]

    # cassandra_cql.cql_query may raise a CommandExecutionError
    try:
        __salt__['cassandra_cql.cql_query_with_prepare'](query,
                                                         'returner_minion',
                                                         tuple(statement_arguments),
                                                         asynchronous=True)
    except CommandExecutionError:
        log.critical('Could not store minion ID with Cassandra returner.')
        raise
    except Exception as e:
        log.critical(
            'Unexpected error while inserting minion ID into the minions '
            'table: %s', e
        )
        raise
32527
function hashpw(password, salt, callback, progress) {
    var real_salt;
    var passwordb = [];
    var saltb = [];
    var hashed = [];
    var minor = String.fromCharCode(0);
    var rounds = 0;
    var off = 0;

    if (!progress) {
        progress = function() {};
    }

    if (salt.charAt(0) != '$' || salt.charAt(1) != '2')
        throw "Invalid salt version";
    if (salt.charAt(2) == '$')
        off = 3;
    else {
        minor = salt.charAt(2);
        if ((minor != 'a' && minor != 'b') || salt.charAt(3) != '$')
            throw "Invalid salt revision";
        off = 4;
    }

    // Extract number of rounds
    if (salt.charAt(off + 2) > '$')
        throw "Missing salt rounds";
    var r1 = parseInt(salt.substring(off, off + 1)) * 10;
    var r2 = parseInt(salt.substring(off + 1, off + 2));
    rounds = r1 + r2;
    real_salt = salt.substring(off + 3, off + 25);
    password = password + (minor >= 'a' ? "\000" : "");
    passwordb = password_to_bytes(password)
    saltb = decode_base64(real_salt, BCRYPT_SALT_LEN);
    crypt_raw(
        passwordb,
        saltb,
        rounds,
        BF_CRYPT_CIPHERTEXT.slice(0),
        function(hashed) {
            var rs = [];
            rs.push("$2");
            if (minor >= 'a')
                rs.push(minor);
            rs.push("$");
            if (rounds < 10)
                rs.push("0");
            rs.push(rounds.toString());
            rs.push("$");
            rs.push(encode_base64(saltb, saltb.length));
            rs.push(encode_base64(hashed, BF_CRYPT_CIPHERTEXT.length * 4 - 1));
            callback(rs.join(''));
        },
        progress);
}
530437
def returner(load):
    '''
    
    '''
    conn = _get_conn()
    if conn is None:
        return None
    cur = conn.cursor()
    sql = '''INSERT INTO salt_returns
            (fun, jid, return, id, success)
            VALUES (%s, %s, %s, %s, %s)'''
    try:
        ret = six.text_type(load['return'])
    except UnicodeDecodeError:
        ret = str(load['return'])
    job_ret = {'return': ret}
    if 'retcode' in load:
        job_ret['retcode'] = load['retcode']
    if 'success' in load:
        job_ret['success'] = load['success']
    cur.execute(
        sql, (
            load['fun'],
            load['jid'],
            salt.utils.json.dumps(job_ret),
            load['id'],
            load.get('success'),
        )
    )
    _close_conn(conn)
876982
def _get_source_sum(source_hash, file_path, saltenv):
    '''
    
    '''
    ret = dict()
    schemes = ('salt', 'http', 'https', 'ftp', 'swift', 's3', 'file')
    invalid_hash_msg = ("Source hash '{0}' format is invalid. It must be in "
                        "the format <hash type>=<hash>").format(source_hash)
    source_hash = six.text_type(source_hash)
    source_hash_scheme = _urlparse(source_hash).scheme

    if source_hash_scheme in schemes:
        # The source_hash is a file on a server
        cached_hash_file = __salt__['cp.cache_file'](source_hash, saltenv)

        if not cached_hash_file:
            raise CommandExecutionError(('Source hash file {0} not'
                                         ' found').format(source_hash))

        ret = __salt__['file.extract_hash'](cached_hash_file, '', file_path)
        if ret is None:
            raise SaltInvocationError(invalid_hash_msg)
    else:
        # The source_hash is a hash string
        items = source_hash.split('=', 1)

        if len(items) != 2:
            invalid_hash_msg = ('{0}, or it must be a supported protocol'
                                ': {1}').format(invalid_hash_msg,
                                                ', '.join(schemes))
            raise SaltInvocationError(invalid_hash_msg)

        ret['hash_type'], ret['hsum'] = [item.strip().lower() for item in items]

    return ret
905553
def returners(opts, functions, whitelist=None, context=None, proxy=None):
    '''
    
    '''
    return LazyLoader(
        _module_dirs(opts, 'returners', 'returner'),
        opts,
        tag='returner',
        whitelist=whitelist,
        pack={'__salt__': functions, '__context__': context, '__proxy__': proxy or {}},
    )
928726
def returner(ret):
    '''
    
    '''
    serv = _get_serv(ret)
    minion = ret['id']
    jid = ret['jid']
    fun = ret['fun']
    rets = salt.utils.json.dumps(ret)
    serv.set('{0}:{1}'.format(jid, minion), rets)  # cache for get_jid
    serv.set('{0}:{1}'.format(fun, minion), rets)  # cache for get_fun

    # The following operations are neither efficient nor atomic.
    # If there is a way to make them so, this should be updated.
    _append_list(serv, 'minions', minion)
    _append_list(serv, 'jids', jid)