golden hour
/opt/saltstack/salt/lib/python3.10/site-packages/salt/utils
⬆️ Go Up
Upload
File/Folder
Size
Actions
__init__.py
237 B
Del
OK
__pycache__
-
Del
OK
aggregation.py
5.17 KB
Del
OK
ansible.py
1.48 KB
Del
OK
args.py
18.33 KB
Del
OK
asynchronous.py
4.06 KB
Del
OK
atomicfile.py
5.33 KB
Del
OK
aws.py
20.37 KB
Del
OK
azurearm.py
11.42 KB
Del
OK
beacons.py
517 B
Del
OK
boto3mod.py
8.36 KB
Del
OK
boto_elb_tag.py
3.02 KB
Del
OK
botomod.py
7.98 KB
Del
OK
cache.py
11.49 KB
Del
OK
channel.py
489 B
Del
OK
cloud.py
116.3 KB
Del
OK
color.py
2.72 KB
Del
OK
compat.py
1.89 KB
Del
OK
configcomparer.py
3.88 KB
Del
OK
configparser.py
10.82 KB
Del
OK
context.py
6.8 KB
Del
OK
crypt.py
5 KB
Del
OK
ctx.py
1.42 KB
Del
OK
data.py
53.03 KB
Del
OK
dateutils.py
2.3 KB
Del
OK
debug.py
4.19 KB
Del
OK
decorators
-
Del
OK
dictdiffer.py
16.48 KB
Del
OK
dicttrim.py
3.9 KB
Del
OK
dictupdate.py
11.33 KB
Del
OK
dns.py
35.21 KB
Del
OK
doc.py
2.25 KB
Del
OK
dockermod
-
Del
OK
entrypoints.py
1.83 KB
Del
OK
environment.py
2.2 KB
Del
OK
error.py
1.18 KB
Del
OK
etcd_util.py
33.25 KB
Del
OK
event.py
52.45 KB
Del
OK
extend.py
8.87 KB
Del
OK
extmods.py
6.04 KB
Del
OK
filebuffer.py
3.15 KB
Del
OK
files.py
27.94 KB
Del
OK
find.py
22.08 KB
Del
OK
fsutils.py
3.29 KB
Del
OK
functools.py
6.02 KB
Del
OK
gitfs.py
130.41 KB
Del
OK
github.py
1.52 KB
Del
OK
gzip_util.py
2.86 KB
Del
OK
hashutils.py
5.91 KB
Del
OK
http.py
33.9 KB
Del
OK
iam.py
1.22 KB
Del
OK
icinga2.py
754 B
Del
OK
idem.py
1.22 KB
Del
OK
immutabletypes.py
2.46 KB
Del
OK
itertools.py
2.36 KB
Del
OK
jid.py
3 KB
Del
OK
jinja.py
33.92 KB
Del
OK
job.py
6.89 KB
Del
OK
json.py
3.78 KB
Del
OK
kickstart.py
41.04 KB
Del
OK
kinds.py
493 B
Del
OK
lazy.py
3.06 KB
Del
OK
listdiffer.py
10.9 KB
Del
OK
locales.py
2.06 KB
Del
OK
mac_utils.py
14.01 KB
Del
OK
mako.py
3.97 KB
Del
OK
master.py
29.7 KB
Del
OK
mattermost.py
1.77 KB
Del
OK
memcached.py
3.56 KB
Del
OK
migrations.py
1.46 KB
Del
OK
mine.py
3.68 KB
Del
OK
minion.py
4.13 KB
Del
OK
minions.py
43.38 KB
Del
OK
mount.py
1.15 KB
Del
OK
msazure.py
5.36 KB
Del
OK
msgpack.py
4.69 KB
Del
OK
nacl.py
13.65 KB
Del
OK
namecheap.py
4.32 KB
Del
OK
napalm.py
23.22 KB
Del
OK
nb_popen.py
7.24 KB
Del
OK
network.py
73.86 KB
Del
OK
nxos.py
12.94 KB
Del
OK
nxos_api.py
4 KB
Del
OK
odict.py
13.21 KB
Del
OK
openstack
-
Del
OK
oset.py
6.41 KB
Del
OK
pagerduty.py
3.03 KB
Del
OK
parsers.py
122.69 KB
Del
OK
path.py
11.24 KB
Del
OK
pbm.py
9.81 KB
Del
OK
pkg
-
Del
OK
platform.py
5.84 KB
Del
OK
powershell.py
4.15 KB
Del
OK
preseed.py
2.64 KB
Del
OK
process.py
40.76 KB
Del
OK
profile.py
3.21 KB
Del
OK
proxy.py
331 B
Del
OK
psutil_compat.py
3.63 KB
Del
OK
pushover.py
4.51 KB
Del
OK
pycrypto.py
5.41 KB
Del
OK
pydsl.py
13.74 KB
Del
OK
pyobjects.py
10.75 KB
Del
OK
reactor.py
18.99 KB
Del
OK
reclass.py
752 B
Del
OK
roster_matcher.py
3.55 KB
Del
OK
rsax931.py
8.42 KB
Del
OK
s3.py
8.78 KB
Del
OK
saltclass.py
14.27 KB
Del
OK
sanitizers.py
2.51 KB
Del
OK
schedule.py
71.81 KB
Del
OK
schema.py
54.26 KB
Del
OK
sdb.py
4.04 KB
Del
OK
slack.py
3.58 KB
Del
OK
smb.py
11.16 KB
Del
OK
smtp.py
3.27 KB
Del
OK
ssdp.py
14.75 KB
Del
OK
ssh.py
769 B
Del
OK
state.py
8.43 KB
Del
OK
stringio.py
355 B
Del
OK
stringutils.py
16.95 KB
Del
OK
systemd.py
5.51 KB
Del
OK
templates.py
24.03 KB
Del
OK
textformat.py
5.03 KB
Del
OK
thin.py
31.91 KB
Del
OK
timed_subprocess.py
4.06 KB
Del
OK
timeout.py
1.53 KB
Del
OK
timeutil.py
2.4 KB
Del
OK
url.py
5 KB
Del
OK
user.py
11.86 KB
Del
OK
validate
-
Del
OK
value.py
247 B
Del
OK
vault.py
21.74 KB
Del
OK
verify.py
25.34 KB
Del
OK
versions.py
17.17 KB
Del
OK
virt.py
3.24 KB
Del
OK
virtualbox.py
22.43 KB
Del
OK
vmware.py
129.74 KB
Del
OK
vsan.py
17.18 KB
Del
OK
vt.py
31.47 KB
Del
OK
vt_helper.py
4.4 KB
Del
OK
win_chcp.py
3.7 KB
Del
OK
win_dacl.py
95.49 KB
Del
OK
win_dotnet.py
4.74 KB
Del
OK
win_functions.py
12.69 KB
Del
OK
win_lgpo_auditpol.py
8.48 KB
Del
OK
win_lgpo_netsh.py
17.87 KB
Del
OK
win_lgpo_reg.py
16.98 KB
Del
OK
win_network.py
16.35 KB
Del
OK
win_osinfo.py
2.83 KB
Del
OK
win_pdh.py
13.85 KB
Del
OK
win_reg.py
30.82 KB
Del
OK
win_runas.py
10.53 KB
Del
OK
win_service.py
5.2 KB
Del
OK
win_system.py
14.47 KB
Del
OK
win_update.py
40.36 KB
Del
OK
winapi.py
818 B
Del
OK
x509.py
73.19 KB
Del
OK
xdg.py
316 B
Del
OK
xmlutil.py
13.91 KB
Del
OK
yaml.py
349 B
Del
OK
yamldumper.py
3.37 KB
Del
OK
yamlencoding.py
1.55 KB
Del
OK
yamllint.py
1.61 KB
Del
OK
yamlloader.py
6.04 KB
Del
OK
yamlloader_old.py
8.15 KB
Del
OK
yast.py
619 B
Del
OK
zeromq.py
1.74 KB
Del
OK
zfs.py
19.15 KB
Del
OK
Edit: master.py
""" salt.utils.master ----------------- Utilities that can only be used on a salt master. """ import logging import os import signal from threading import Event, Thread import salt.cache import salt.client import salt.config import salt.payload import salt.pillar import salt.utils.atomicfile import salt.utils.files import salt.utils.minions import salt.utils.platform import salt.utils.stringutils import salt.utils.verify from salt.exceptions import SaltException from salt.utils.cache import CacheCli as cache_cli from salt.utils.process import Process from salt.utils.zeromq import zmq log = logging.getLogger(__name__) def get_running_jobs(opts): """ Return the running jobs on this minion """ ret = [] proc_dir = os.path.join(opts["cachedir"], "proc") if not os.path.isdir(proc_dir): return ret for fn_ in os.listdir(proc_dir): path = os.path.join(proc_dir, fn_) try: data = _read_proc_file(path, opts) if data is not None: ret.append(data) except OSError: # proc files may be removed at any time during this process by # the master process that is executing the JID in question, so # we must ignore ENOENT during this process log.trace("%s removed during processing by master process", path) return ret def _read_proc_file(path, opts): """ Return a dict of JID metadata, or None """ with salt.utils.files.fopen(path, "rb") as fp_: buf = fp_.read() fp_.close() if buf: data = salt.payload.loads(buf) else: # Proc file is empty, remove try: os.remove(path) except OSError: log.debug("Unable to remove proc file %s.", path) return None if not isinstance(data, dict): # Invalid serial object return None if not salt.utils.process.os_is_running(data["pid"]): # The process is no longer running, clear out the file and # continue try: os.remove(path) except OSError: log.debug("Unable to remove proc file %s.", path) return None if not _check_cmdline(data): pid = data.get("pid") if pid: log.warning("PID %s exists but does not appear to be a salt process.", pid) try: os.remove(path) except OSError: log.debug("Unable to remove proc file %s.", path) return None return data def _check_cmdline(data): """ In some cases where there are an insane number of processes being created on a system a PID can get recycled or assigned to a non-Salt process. On Linux this fn checks to make sure the PID we are checking on is actually a Salt process. For non-Linux systems we punt and just return True """ if not salt.utils.platform.is_linux(): return True pid = data.get("pid") if not pid: return False if not os.path.isdir("/proc"): return True path = os.path.join("/proc/{}/cmdline".format(pid)) if not os.path.isfile(path): return False try: with salt.utils.files.fopen(path, "rb") as fp_: return b"salt" in fp_.read() except OSError: return False class MasterPillarUtil: """ Helper utility for easy access to targeted minion grain and pillar data, either from cached data on the master or retrieved on demand, or (by default) both. The minion pillar data returned in get_minion_pillar() is compiled directly from salt.pillar.Pillar on the master to avoid any possible 'pillar poisoning' from a compromised or untrusted minion. ** However, the minion grains are still possibly entirely supplied by the minion. ** Example use case: For runner modules that need access minion pillar data, MasterPillarUtil.get_minion_pillar should be used instead of getting the pillar data by executing the "pillar" module on the minions: # my_runner.py tgt = 'web*' pillar_util = salt.utils.master.MasterPillarUtil(tgt, tgt_type='glob', opts=__opts__) pillar_data = pillar_util.get_minion_pillar() """ def __init__( self, tgt="", tgt_type="glob", saltenv=None, use_cached_grains=True, use_cached_pillar=True, grains_fallback=True, pillar_fallback=True, opts=None, ): log.debug("New instance of %s created.", self.__class__.__name__) if opts is None: log.error("%s: Missing master opts init arg.", self.__class__.__name__) raise SaltException( "{}: Missing master opts init arg.".format(self.__class__.__name__) ) else: self.opts = opts self.tgt = tgt self.tgt_type = tgt_type self.saltenv = saltenv self.use_cached_grains = use_cached_grains self.use_cached_pillar = use_cached_pillar self.grains_fallback = grains_fallback self.pillar_fallback = pillar_fallback self.cache = salt.cache.factory(opts) log.debug( "Init settings: tgt: '%s', tgt_type: '%s', saltenv: '%s', " "use_cached_grains: %s, use_cached_pillar: %s, " "grains_fallback: %s, pillar_fallback: %s", tgt, tgt_type, saltenv, use_cached_grains, use_cached_pillar, grains_fallback, pillar_fallback, ) def _get_cached_mine_data(self, *minion_ids): # Return one dict with the cached mine data of the targeted minions mine_data = {minion_id: {} for minion_id in minion_ids} if not self.opts.get("minion_data_cache", False) and not self.opts.get( "enforce_mine_cache", False ): log.debug( "Skipping cached mine data minion_data_cache" "and enfore_mine_cache are both disabled." ) return mine_data if not minion_ids: minion_ids = self.cache.list("minions") for minion_id in minion_ids: if not salt.utils.verify.valid_id(self.opts, minion_id): continue mdata = self.cache.fetch("minions/{}".format(minion_id), "mine") if isinstance(mdata, dict): mine_data[minion_id] = mdata return mine_data def _get_cached_minion_data(self, *minion_ids): # Return two separate dicts of cached grains and pillar data of the # minions grains = {minion_id: {} for minion_id in minion_ids} pillars = grains.copy() if not self.opts.get("minion_data_cache", False): log.debug("Skipping cached data because minion_data_cache is not enabled.") return grains, pillars if not minion_ids: minion_ids = self.cache.list("minions") for minion_id in minion_ids: if not salt.utils.verify.valid_id(self.opts, minion_id): continue mdata = self.cache.fetch("minions/{}".format(minion_id), "data") if not isinstance(mdata, dict): log.warning( "cache.fetch should always return a dict. ReturnedType: %s," " MinionId: %s", type(mdata).__name__, minion_id, ) continue if "grains" in mdata: grains[minion_id] = mdata["grains"] if "pillar" in mdata: pillars[minion_id] = mdata["pillar"] return grains, pillars def _get_live_minion_grains(self, minion_ids): # Returns a dict of grains fetched directly from the minions log.debug('Getting live grains for minions: "%s"', minion_ids) with salt.client.get_local_client(self.opts["conf_file"]) as client: return client.cmd( ",".join(minion_ids), "grains.items", timeout=self.opts["timeout"], tgt_type="list", ) def _get_live_minion_pillar(self, minion_id=None, minion_grains=None): # Returns a dict of pillar data for one minion if minion_id is None: return {} if not minion_grains: log.warning("Cannot get pillar data for %s: no grains supplied.", minion_id) return {} log.debug("Getting live pillar for %s", minion_id) pillar = salt.pillar.Pillar( self.opts, minion_grains, minion_id, self.saltenv, self.opts["ext_pillar"] ) log.debug("Compiling pillar for %s", minion_id) ret = pillar.compile_pillar() return ret def _get_minion_grains(self, *minion_ids, **kwargs): # Get the minion grains either from cache or from a direct query # on the minion. By default try to use cached grains first, then # fall back to querying the minion directly. ret = {} cached_grains = kwargs.get("cached_grains", {}) cret = {} lret = {} if self.use_cached_grains: cret = { minion_id: mcache for (minion_id, mcache) in cached_grains.items() if mcache } missed_minions = [ minion_id for minion_id in minion_ids if minion_id not in cret ] log.debug("Missed cached minion grains for: %s", missed_minions) if self.grains_fallback: lret = self._get_live_minion_grains(missed_minions) ret = dict( list({minion_id: {} for minion_id in minion_ids}.items()) + list(lret.items()) + list(cret.items()) ) else: lret = self._get_live_minion_grains(minion_ids) missed_minions = [ minion_id for minion_id in minion_ids if minion_id not in lret ] log.debug("Missed live minion grains for: %s", missed_minions) if self.grains_fallback: cret = { minion_id: mcache for (minion_id, mcache) in cached_grains.items() if mcache } ret = dict( list({minion_id: {} for minion_id in minion_ids}.items()) + list(lret.items()) + list(cret.items()) ) return ret def _get_minion_pillar(self, *minion_ids, **kwargs): # Get the minion pillar either from cache or from a direct query # on the minion. By default try use the cached pillar first, then # fall back to rendering pillar on demand with the supplied grains. ret = {} grains = kwargs.get("grains", {}) cached_pillar = kwargs.get("cached_pillar", {}) cret = {} lret = {} if self.use_cached_pillar: cret = { minion_id: mcache for (minion_id, mcache) in cached_pillar.items() if mcache } missed_minions = [ minion_id for minion_id in minion_ids if minion_id not in cret ] log.debug("Missed cached minion pillars for: %s", missed_minions) if self.pillar_fallback: lret = { minion_id: self._get_live_minion_pillar( minion_id, grains.get(minion_id, {}) ) for minion_id in missed_minions } ret = dict( list({minion_id: {} for minion_id in minion_ids}.items()) + list(lret.items()) + list(cret.items()) ) else: lret = { minion_id: self._get_live_minion_pillar( minion_id, grains.get(minion_id, {}) ) for minion_id in minion_ids } missed_minions = [ minion_id for minion_id in minion_ids if minion_id not in lret ] log.debug("Missed live minion pillars for: %s", missed_minions) if self.pillar_fallback: cret = { minion_id: mcache for (minion_id, mcache) in cached_pillar.items() if mcache } ret = dict( list({minion_id: {} for minion_id in minion_ids}.items()) + list(lret.items()) + list(cret.items()) ) return ret def _tgt_to_list(self): # Return a list of minion ids that match the target and tgt_type minion_ids = [] ckminions = salt.utils.minions.CkMinions(self.opts) _res = ckminions.check_minions(self.tgt, self.tgt_type) minion_ids = _res["minions"] if not minion_ids: log.debug( 'No minions matched for tgt="%s" and tgt_type="%s"', self.tgt, self.tgt_type, ) return {} log.debug( 'Matching minions for tgt="%s" and tgt_type="%s": %s', self.tgt, self.tgt_type, minion_ids, ) return minion_ids def get_minion_pillar(self): """ Get pillar data for the targeted minions, either by fetching the cached minion data on the master, or by compiling the minion's pillar data on the master. For runner modules that need access minion pillar data, this function should be used instead of getting the pillar data by executing the pillar module on the minions. By default, this function tries hard to get the pillar data: - Try to get the cached minion grains and pillar if the master has minion_data_cache: True - If the pillar data for the minion is cached, use it. - If there is no cached grains/pillar data for a minion, then try to get the minion grains directly from the minion. - Use the minion grains to compile the pillar directly from the master using salt.pillar.Pillar """ minion_pillars = {} minion_grains = {} minion_ids = self._tgt_to_list() if self.tgt and not minion_ids: return {} if any( arg for arg in [ self.use_cached_grains, self.use_cached_pillar, self.grains_fallback, self.pillar_fallback, ] ): log.debug("Getting cached minion data") cached_minion_grains, cached_minion_pillars = self._get_cached_minion_data( *minion_ids ) else: cached_minion_grains = {} cached_minion_pillars = {} log.debug("Getting minion grain data for: %s", minion_ids) minion_grains = self._get_minion_grains( *minion_ids, cached_grains=cached_minion_grains ) log.debug("Getting minion pillar data for: %s", minion_ids) minion_pillars = self._get_minion_pillar( *minion_ids, grains=minion_grains, cached_pillar=cached_minion_pillars ) return minion_pillars def get_minion_grains(self): """ Get grains data for the targeted minions, either by fetching the cached minion data on the master, or by fetching the grains directly on the minion. By default, this function tries hard to get the grains data: - Try to get the cached minion grains if the master has minion_data_cache: True - If the grains data for the minion is cached, use it. - If there is no cached grains data for a minion, then try to get the minion grains directly from the minion. """ minion_grains = {} minion_ids = self._tgt_to_list() if not minion_ids: return {} if any(arg for arg in [self.use_cached_grains, self.grains_fallback]): log.debug("Getting cached minion data.") cached_minion_grains, cached_minion_pillars = self._get_cached_minion_data( *minion_ids ) else: cached_minion_grains = {} log.debug("Getting minion grain data for: %s", minion_ids) minion_grains = self._get_minion_grains( *minion_ids, cached_grains=cached_minion_grains ) return minion_grains def get_cached_mine_data(self): """ Get cached mine data for the targeted minions. """ mine_data = {} minion_ids = self._tgt_to_list() log.debug("Getting cached mine data for: %s", minion_ids) mine_data = self._get_cached_mine_data(*minion_ids) return mine_data def clear_cached_minion_data( self, clear_pillar=False, clear_grains=False, clear_mine=False, clear_mine_func=None, ): """ Clear the cached data/files for the targeted minions. """ clear_what = [] if clear_pillar: clear_what.append("pillar") if clear_grains: clear_what.append("grains") if clear_mine: clear_what.append("mine") if clear_mine_func is not None: clear_what.append("mine_func: '{}'".format(clear_mine_func)) if not clear_what: log.debug("No cached data types specified for clearing.") return False minion_ids = self._tgt_to_list() log.debug("Clearing cached %s data for: %s", ", ".join(clear_what), minion_ids) if clear_pillar == clear_grains: # clear_pillar and clear_grains are both True or both False. # This means we don't deal with pillar/grains caches at all. grains = {} pillars = {} else: # Unless both clear_pillar and clear_grains are True, we need # to read in the pillar/grains data since they are both stored # in the same file, 'data.p' grains, pillars = self._get_cached_minion_data(*minion_ids) try: c_minions = self.cache.list("minions") for minion_id in minion_ids: if not salt.utils.verify.valid_id(self.opts, minion_id): continue if minion_id not in c_minions: # Cache bank for this minion does not exist. Nothing to do. continue bank = "minions/{}".format(minion_id) minion_pillar = pillars.pop(minion_id, False) minion_grains = grains.pop(minion_id, False) if ( (clear_pillar and clear_grains) or (clear_pillar and not minion_grains) or (clear_grains and not minion_pillar) ): # Not saving pillar or grains, so just delete the cache file self.cache.flush(bank, "data") elif clear_pillar and minion_grains: self.cache.store(bank, "data", {"grains": minion_grains}) elif clear_grains and minion_pillar: self.cache.store(bank, "data", {"pillar": minion_pillar}) if clear_mine: # Delete the whole mine file self.cache.flush(bank, "mine") elif clear_mine_func is not None: # Delete a specific function from the mine file mine_data = self.cache.fetch(bank, "mine") if isinstance(mine_data, dict): if mine_data.pop(clear_mine_func, False): self.cache.store(bank, "mine", mine_data) except OSError: return True return True class CacheTimer(Thread): """ A basic timer class the fires timer-events every second. This is used for cleanup by the ConnectedCache() """ def __init__(self, opts, event): Thread.__init__(self) self.opts = opts self.stopped = event self.daemon = True self.timer_sock = os.path.join(self.opts["sock_dir"], "con_timer.ipc") def run(self): """ main loop that fires the event every second """ context = zmq.Context() # the socket for outgoing timer events socket = context.socket(zmq.PUB) socket.setsockopt(zmq.LINGER, 100) socket.bind("ipc://" + self.timer_sock) count = 0 log.debug("ConCache-Timer started") while not self.stopped.wait(1): socket.send(salt.payload.dumps(count)) # pylint: disable=missing-kwoa count += 1 if count >= 60: count = 0 class CacheWorker(Process): """ Worker for ConnectedCache which runs in its own process to prevent blocking of ConnectedCache main-loop when refreshing minion-list """ def __init__(self, opts, **kwargs): """ Sets up the zmq-connection to the ConCache """ super().__init__(**kwargs) self.opts = opts def run(self): """ Gather currently connected minions and update the cache """ new_mins = list(salt.utils.minions.CkMinions(self.opts).connected_ids()) cc = cache_cli(self.opts) cc.get_cached() cc.put_cache([new_mins]) log.debug("ConCache CacheWorker update finished") class ConnectedCache(Process): """ Provides access to all minions ids that the master has successfully authenticated. The cache is cleaned up regularly by comparing it to the IPs that have open connections to the master publisher port. """ def __init__(self, opts, **kwargs): """ starts the timer and inits the cache itself """ super().__init__(**kwargs) log.debug("ConCache initializing...") # the possible settings for the cache self.opts = opts # the actual cached minion ids self.minions = [] self.cache_sock = os.path.join(self.opts["sock_dir"], "con_cache.ipc") self.update_sock = os.path.join(self.opts["sock_dir"], "con_upd.ipc") self.upd_t_sock = os.path.join(self.opts["sock_dir"], "con_timer.ipc") self.cleanup() # the timer provides 1-second intervals to the loop in run() # to make the cache system most responsive, we do not use a loop- # delay which makes it hard to get 1-second intervals without a timer self.timer_stop = Event() self.timer = CacheTimer(self.opts, self.timer_stop) self.timer.start() self.running = True def signal_handler(self, sig, frame): """ handle signals and shutdown """ self.stop() def cleanup(self): """ remove sockets on shutdown """ log.debug("ConCache cleaning up") if os.path.exists(self.cache_sock): os.remove(self.cache_sock) if os.path.exists(self.update_sock): os.remove(self.update_sock) if os.path.exists(self.upd_t_sock): os.remove(self.upd_t_sock) def secure(self): """ secure the sockets for root-only access """ log.debug("ConCache securing sockets") if os.path.exists(self.cache_sock): os.chmod(self.cache_sock, 0o600) if os.path.exists(self.update_sock): os.chmod(self.update_sock, 0o600) if os.path.exists(self.upd_t_sock): os.chmod(self.upd_t_sock, 0o600) def stop(self): """ shutdown cache process """ # avoid getting called twice self.cleanup() if self.running: self.running = False self.timer_stop.set() self.timer.join() def run(self): """ Main loop of the ConCache, starts updates in intervals and answers requests from the MWorkers """ context = zmq.Context() # the socket for incoming cache requests creq_in = context.socket(zmq.REP) creq_in.setsockopt(zmq.LINGER, 100) creq_in.bind("ipc://" + self.cache_sock) # the socket for incoming cache-updates from workers cupd_in = context.socket(zmq.SUB) cupd_in.setsockopt(zmq.SUBSCRIBE, b"") cupd_in.setsockopt(zmq.LINGER, 100) cupd_in.bind("ipc://" + self.update_sock) # the socket for the timer-event timer_in = context.socket(zmq.SUB) timer_in.setsockopt(zmq.SUBSCRIBE, b"") timer_in.setsockopt(zmq.LINGER, 100) timer_in.connect("ipc://" + self.upd_t_sock) poller = zmq.Poller() poller.register(creq_in, zmq.POLLIN) poller.register(cupd_in, zmq.POLLIN) poller.register(timer_in, zmq.POLLIN) # register a signal handler signal.signal(signal.SIGINT, self.signal_handler) # secure the sockets from the world self.secure() log.info("ConCache started") while self.running: # we check for new events with the poller try: socks = dict(poller.poll(1)) except KeyboardInterrupt: self.stop() except zmq.ZMQError as zmq_err: log.error("ConCache ZeroMQ-Error occurred") log.exception(zmq_err) self.stop() # check for next cache-request if socks.get(creq_in) == zmq.POLLIN: msg = salt.payload.loads(creq_in.recv()) log.debug("ConCache Received request: %s", msg) # requests to the minion list are send as str's if isinstance(msg, str): if msg == "minions": # Send reply back to client reply = salt.payload.dumps(self.minions) creq_in.send(reply) # pylint: disable=missing-kwoa # check for next cache-update from workers if socks.get(cupd_in) == zmq.POLLIN: new_c_data = salt.payload.loads(cupd_in.recv()) # tell the worker to exit # cupd_in.send(serial.dumps('ACK')) # check if the returned data is usable if not isinstance(new_c_data, list): log.error("ConCache Worker returned unusable result") del new_c_data continue # the cache will receive lists of minions # 1. if the list only has 1 item, its from an MWorker, we append it # 2. if the list contains another list, its from a CacheWorker and # the currently cached minions are replaced with that list # 3. anything else is considered malformed try: if not new_c_data: log.debug("ConCache Got empty update from worker") continue data = new_c_data[0] if isinstance(data, str): if data not in self.minions: log.debug( "ConCache Adding minion %s to cache", new_c_data[0] ) self.minions.append(data) elif isinstance(data, list): log.debug("ConCache Replacing minion list from worker") self.minions = data except IndexError: log.debug("ConCache Got malformed result dict from worker") del new_c_data log.info("ConCache %s entries in cache", len(self.minions)) # check for next timer-event to start new jobs if socks.get(timer_in) == zmq.POLLIN: sec_event = salt.payload.loads(timer_in.recv()) # update the list every 30 seconds if int(sec_event % 30) == 0: cw = CacheWorker(self.opts) cw.start() self.stop() creq_in.close() cupd_in.close() timer_in.close() context.term() log.debug("ConCache Shutting down") def ping_all_connected_minions(opts): """ Ping all connected minions. """ if opts["minion_data_cache"]: tgt = list(salt.utils.minions.CkMinions(opts).connected_ids()) form = "list" else: tgt = "*" form = "glob" with salt.client.LocalClient() as client: client.cmd_async(tgt, "test.ping", tgt_type=form) def get_master_key(key_user, opts, skip_perm_errors=False): """ Return the master key. """ if key_user == "root": if opts.get("user", "root") != "root": key_user = opts.get("user", "root") if key_user.startswith("sudo_"): key_user = opts.get("user", "root") if salt.utils.platform.is_windows(): # The username may contain '\' if it is in Windows # 'DOMAIN\username' format. Fix this for the keyfile path. key_user = key_user.replace("\\", "_") keyfile = os.path.join(opts["cachedir"], ".{}_key".format(key_user)) # Make sure all key parent directories are accessible salt.utils.verify.check_path_traversal(opts["cachedir"], key_user, skip_perm_errors) try: with salt.utils.files.fopen(keyfile, "r") as key: return key.read() except OSError: # Fall back to eauth return "" def get_values_of_matching_keys(pattern_dict, user_name): """ Check a whitelist and/or blacklist to see if the value matches it. """ ret = [] for expr in pattern_dict: if salt.utils.stringutils.expr_match(user_name, expr): ret.extend(pattern_dict[expr]) return ret # test code for the ConCache class if __name__ == "__main__": opts = salt.config.master_config("/etc/salt/master") conc = ConnectedCache(opts) conc.start()
Save