xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-08-05 10:17:51 +00:00
Code Issues Projects Releases Wiki Activity

Merge remote-tracking branch 'xcat/master'

Browse Source
This commit is contained in:
Jarrod Johnson
2026-07-15 14:47:51 -04:00
parent 164a168694 ea9be4aae0
commit a0995b63ba
11 changed files with 234 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+64
View File
@@ -0,0 +1,64 @@
name: CI
on:
push:
pull_request:
permissions:
contents: read
jobs:
ShellCheck:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- name: Run ShellCheck (errors only)
# Check every tracked file that has a .sh extension or an sh/bash
# shebang. SC2148 (missing shebang) is excluded because many .sh
# files are sourced fragments or dracut hooks; ShellCheck then
# falls back to checking them as bash.
run: |
{
git ls-files '*.sh'
git ls-files | while IFS= read -r f; do
[ -f "$f" ] || continue
head -c 200 "$f" | head -n 1 | \
grep -qE '^#!.*[/ ](sh|bash|dash|ash|ksh)([ \t]|$)' && echo "$f"
done
} | sort -u | xargs -d '\n' shellcheck --severity=error --exclude=SC2148
python-compileall:
name: Python compileall
runs-on: ubuntu-latest
env:
# One entry per Python version shipped by the distros confluent
# targets, limited to versions actions/setup-python still provides
# on current runners (sles15/alma8 ship 3.6, which is unavailable).
# Newline-separated so it feeds both setup-python (multiline input)
# and the shell loop below (word-split on whitespace).
PYTHON_VERSIONS: |
3.8
3.9
3.10
3.12
3.13
3.14
steps:
- uses: actions/checkout@v7
- uses: actions/setup-python@v6
with:
python-version: ${{ env.PYTHON_VERSIONS }}
- name: Compile all Python files
run: |
rc=0
for v in $PYTHON_VERSIONS; do
echo "::group::Python $v"
if "python$v" -W error -m compileall -q -x '/\.git/' .; then
echo "::endgroup::"
else
echo "::endgroup::"
echo "::error::Python $v compileall failed"
rc=1
fi
done
exit "$rc"
confluent_osdeploy/el7-diskless/profiles/default/scripts/image2disk.py
+3 -3
View File
@@ -173,8 +173,8 @@ def fixup(rootdir, vols):
for vol in vols:
if vol['mount'] == '/boot/efi':
targdev = vol['targetdisk']
partnum = re.search('(\d+)$', targdev).group(1)
targblock = re.search('(.*)\d+$', targdev).group(1)
partnum = re.search(r'(\d+)$', targdev).group(1)
targblock = re.search(r'(.*)\d+$', targdev).group(1)
if targblock:
if targblock.endswith('p') and 'nvme' in targblock:
targblock = targblock[:-1]
@@ -231,7 +231,7 @@ def install_to_disk(imgpath):
deflvmsize += fs['initsize']
minlvmsize += fs['minsize']
else:
plainvols[int(re.search('(\d+)$', fs['device'])[0])] = fs
plainvols[int(re.search(r'(\d+)$', fs['device'])[0])] = fs
with open('/tmp/installdisk') as diskin:
instdisk = diskin.read()
instdisk = '/dev/' + instdisk
confluent_osdeploy/ubuntu20.04-diskless/profiles/default/scripts/image2disk.py
+3 -3
View File
@@ -204,8 +204,8 @@ def fixup(rootdir, vols):
for vol in vols:
if vol['mount'] == '/boot/efi':
targdev = vol['targetdisk']
partnum = re.search('(\d+)$', targdev).group(1)
targblock = re.search('(.*)\d+$', targdev).group(1)
partnum = re.search(r'(\d+)$', targdev).group(1)
targblock = re.search(r'(.*)\d+$', targdev).group(1)
if targblock:
if targblock.endswith('p') and 'nvme' in targblock:
targblock = targblock[:-1]
@@ -263,7 +263,7 @@ def install_to_disk(imgpath):
deflvmsize += fs['initsize']
minlvmsize += fs['minsize']
else:
plainvols[int(re.search('(\d+)$', fs['device'])[0])] = fs
plainvols[int(re.search(r'(\d+)$', fs['device'])[0])] = fs
with open('/tmp/installdisk') as diskin:
instdisk = diskin.read()
instdisk = '/dev/' + instdisk
confluent_server/builddeb
+22 -4
View File
@@ -50,12 +50,30 @@ if ! grep wheezy /etc/os-release; then
fi
head -n -1 debian/control > debian/control1
mv debian/control1 debian/control
cat > debian/postinst << EOF
if ! getent passwd confluent > /dev/null; then
cat > debian/postinst << \EOF
if ! getent passwd confluent > /dev/null; then
useradd -r confluent -d /var/lib/confluent -s /usr/sbin/nologin
mkdir -p /etc/confluent
chown confluent /etc/confluent
fi
mkdir -p /etc/confluent /var/lib/confluent /var/log/confluent /var/cache/confluent
chown confluent:confluent /etc/confluent /var/lib/confluent /var/log/confluent /var/cache/confluent
sysctl -p /usr/lib/sysctl.d/confluent.conf > /dev/null 2>&1
NEEDCHOWN=0
NEEDSTART=0
[ -n "$(find /etc/confluent /var/log/confluent /var/cache/confluent -uid 0 -print -quit 2>/dev/null)" ] && NEEDCHOWN=1
if [ $NEEDCHOWN = 1 ]; then
if systemctl is-active confluent > /dev/null; then
NEEDSTART=1
systemctl stop confluent
fi
chown -R confluent:confluent /etc/confluent /var/log/confluent /var/cache/confluent
fi
systemctl daemon-reload
if systemctl is-active confluent > /dev/null || [ $NEEDSTART = 1 ]; then systemctl restart confluent > /dev/null 2>&1; fi
if [ ! -e /etc/pam.d/confluent ]; then
ln -s /etc/pam.d/sshd /etc/pam.d/confluent
fi
true
EOF
echo 'export PYBUILD_INSTALL_ARGS=--install-lib=/opt/confluent/lib/python' >> debian/rules
#echo 'Provides: python-'$DPKGNAME >> debian/control
confluent_server/confluent/certutil.py
+109 -78
View File
@@ -207,85 +207,98 @@ def substitute_cfg(setting, key, val, newval, cfgfile, line):
return False
async def create_full_ca(certout):
mkdirp('/etc/confluent/tls/ca/private')
keyout = '/etc/confluent/tls/ca/private/cakey.pem'
csrout = '/etc/confluent/tls/ca/ca.csr'
mkdirp('/etc/confluent/tls/ca/newcerts')
with open('/etc/confluent/tls/ca/index.txt', 'w') as idx:
pass
with open('/etc/confluent/tls/ca/index.txt.attr', 'w') as idx:
idx.write('unique_subject = no')
with open('/etc/confluent/tls/ca/serial', 'w') as srl:
srl.write('01')
sslcfg = get_openssl_conf_location()
newcfg = '/etc/confluent/tls/ca/openssl.cfg'
settings = {
'dir': '/etc/confluent/tls/ca',
'certificate': '$dir/cacert.pem',
'private_key': '$dir/private/cakey.pem',
'countryName': 'optional',
'stateOrProvinceName': 'optional',
'organizationName': 'optional',
}
subj = '/CN=Confluent TLS Certificate authority ({0})'.format(socket.gethostname())
if len(subj) > 68:
subj = subj[:68]
with open(sslcfg, 'r') as cfgin:
with open(newcfg, 'w') as cfgfile:
for line in cfgin.readlines():
cfg = line.split('#')[0]
if '=' in cfg:
key, val = cfg.split('=', 1)
for stg in settings:
if substitute_cfg(stg, key, val, settings[stg], cfgfile, line):
break
else:
cfgfile.write(line.strip() + '\n')
continue
cfgfile.write(line.strip() + '\n')
cfgfile.write('\n[CACert]\nbasicConstraints = critical,CA:true\nkeyUsage = critical,keyCertSign,cRLSign\n[ca_confluent]\n')
await util.check_call(
'openssl', 'ecparam', '-name', 'secp384r1', '-genkey', '-out',
keyout)
await util.check_call(
'openssl', 'req', '-new', '-key', keyout, '-out', csrout, '-subj', subj)
await util.check_call(
'openssl', 'ca', '-config', newcfg, '-batch', '-selfsign',
'-extensions', 'CACert', '-extfile', newcfg,
'-notext', '-md', 'sha384', '-startdate',
'19700101010101Z', '-enddate', '21000101010101Z', '-keyfile',
keyout, '-out', '/etc/confluent/tls/ca/cacert.pem', '-in', csrout
)
shutil.copy2('/etc/confluent/tls/ca/cacert.pem', certout)
# The CA is used by the confluent service, which runs as the owner of
# /etc/confluent rather than root; create the CA material as that user
# so the service can use the database for issuing certificates
ouid = normalize_uid()
try:
mkdirp('/etc/confluent/tls/ca/private')
keyout = '/etc/confluent/tls/ca/private/cakey.pem'
csrout = '/etc/confluent/tls/ca/ca.csr'
mkdirp('/etc/confluent/tls/ca/newcerts')
with open('/etc/confluent/tls/ca/index.txt', 'w') as idx:
pass
with open('/etc/confluent/tls/ca/index.txt.attr', 'w') as idx:
idx.write('unique_subject = no')
with open('/etc/confluent/tls/ca/serial', 'w') as srl:
srl.write('01')
sslcfg = get_openssl_conf_location()
newcfg = '/etc/confluent/tls/ca/openssl.cfg'
settings = {
'dir': '/etc/confluent/tls/ca',
'certificate': '$dir/cacert.pem',
'private_key': '$dir/private/cakey.pem',
'countryName': 'optional',
'stateOrProvinceName': 'optional',
'organizationName': 'optional',
}
subj = '/CN=Confluent TLS Certificate authority ({0})'.format(socket.gethostname())
if len(subj) > 68:
subj = subj[:68]
with open(sslcfg, 'r') as cfgin:
with open(newcfg, 'w') as cfgfile:
for line in cfgin.readlines():
cfg = line.split('#')[0]
if '=' in cfg:
key, val = cfg.split('=', 1)
for stg in settings:
if substitute_cfg(stg, key, val, settings[stg], cfgfile, line):
break
else:
cfgfile.write(line.strip() + '\n')
continue
cfgfile.write(line.strip() + '\n')
cfgfile.write('\n[CACert]\nbasicConstraints = critical,CA:true\nkeyUsage = critical,keyCertSign,cRLSign\n[ca_confluent]\n')
await util.check_call(
'openssl', 'ecparam', '-name', 'secp384r1', '-genkey', '-out',
keyout)
await util.check_call(
'openssl', 'req', '-new', '-key', keyout, '-out', csrout, '-subj', subj)
await util.check_call(
'openssl', 'ca', '-config', newcfg, '-batch', '-selfsign',
'-extensions', 'CACert', '-extfile', newcfg,
'-notext', '-md', 'sha384', '-startdate',
'19700101010101Z', '-enddate', '21000101010101Z', '-keyfile',
keyout, '-out', '/etc/confluent/tls/ca/cacert.pem', '-in', csrout
)
shutil.copy2('/etc/confluent/tls/ca/cacert.pem', certout)
finally:
os.seteuid(ouid)
#openssl ca -config openssl.cnf -selfsign -keyfile cakey.pem -startdate 20150214120000Z -enddate 20160214120000Z
#20160107071311Z -enddate 20170106071311Z
async def create_simple_ca(keyout, certout):
# As with create_full_ca, the CA material must be owned by the owner
# of /etc/confluent for use by the confluent service
ouid = normalize_uid()
try:
os.makedirs('/etc/confluent/tls')
except OSError as e:
if e.errno != 17:
raise
sslcfg = get_openssl_conf_location()
tmphdl, tmpconfig = tempfile.mkstemp()
os.close(tmphdl)
shutil.copy2(sslcfg, tmpconfig)
await util.check_call(
'openssl', 'ecparam', '-name', 'secp384r1', '-genkey', '-out',
keyout)
try:
subj = '/CN=Confluent TLS Certificate authority ({0})'.format(socket.gethostname())
if len(subj) > 68:
subj = subj[:68]
with open(tmpconfig, 'a') as cfgfile:
cfgfile.write('\n[CACert]\nbasicConstraints = critical,CA:true\n')
try:
os.makedirs('/etc/confluent/tls')
except OSError as e:
if e.errno != 17:
raise
sslcfg = get_openssl_conf_location()
tmphdl, tmpconfig = tempfile.mkstemp()
os.close(tmphdl)
shutil.copy2(sslcfg, tmpconfig)
await util.check_call(
'openssl', 'req', '-new', '-x509', '-key', keyout, '-days',
'27300', '-out', certout, '-subj', subj,
'-extensions', 'CACert', '-config', tmpconfig
)
'openssl', 'ecparam', '-name', 'secp384r1', '-genkey', '-out',
keyout)
try:
subj = '/CN=Confluent TLS Certificate authority ({0})'.format(socket.gethostname())
if len(subj) > 68:
subj = subj[:68]
with open(tmpconfig, 'a') as cfgfile:
cfgfile.write('\n[CACert]\nbasicConstraints = critical,CA:true\n')
await util.check_call(
'openssl', 'req', '-new', '-x509', '-key', keyout, '-days',
'27300', '-out', certout, '-subj', subj,
'-extensions', 'CACert', '-config', tmpconfig
)
finally:
os.remove(tmpconfig)
finally:
os.remove(tmpconfig)
os.seteuid(ouid)
async def create_certificate(keyout=None, certout=None, csrfile=None, subj=None, san=None, backdate=True, days=None):
now_utc = datetime.datetime.now(datetime.timezone.utc)
@@ -413,12 +426,30 @@ async def create_certificate(keyout=None, certout=None, csrfile=None, subj=None,
shutil.copy2(cacfgfile, tmpcafile)
os.close(tmphdl)
cacfgfile = tmpcafile
await util.check_call(
'openssl', 'ca', '-config', cacfgfile, '-rand_serial',
'-in', csrfile, '-out', certout, '-batch', '-notext',
'-startdate', startdate, '-enddate', enddate, '-md', 'sha384',
'-extfile', extconfig, '-subj', subj
)
os.chmod(cacfgfile, 0o644)
os.chmod(csrfile, 0o644)
# openssl ca rewrites the CA database (index, serial) as the
# invoking user; run it as the owner of /etc/confluent so the
# database remains usable by the confluent service. The chmodded
# temporary inputs hold no secrets, and the certificate is
# written to a temporary path first, as certout may only be
# writable by the original user (e.g. a web server certificate
# path during osdeploy initialize -t)
os.chmod(extconfig, 0o644)
ouid = normalize_uid()
try:
tmphdl, tmpcertout = tempfile.mkstemp()
os.close(tmphdl)
await util.check_call(
'openssl', 'ca', '-config', cacfgfile, '-rand_serial',
'-in', csrfile, '-out', tmpcertout, '-batch', '-notext',
'-startdate', startdate, '-enddate', enddate, '-md', 'sha384',
'-extfile', extconfig, '-subj', subj
)
finally:
os.seteuid(ouid)
shutil.copy(tmpcertout, certout)
os.remove(tmpcertout)
for keycopy in tlsmateriallocation.get('keys', []):
if keycopy != keyout:
shutil.copy2(keyout, keycopy)
confluent_server/confluent/collective/manager.py
+14 -12
View File
@@ -206,28 +206,30 @@ async def follow_leader(remote, leader):
exitcause = await cfm.follow_channel(remote)
newleader = exitcause.get('newleader', None)
finally:
handled = False
if cleanexit:
log.log({'info': 'Previous following cleanly closed',
'subsystem': 'collective'})
return
if newleader:
handled = True
if not handled and newleader:
log.log(
{'info': 'Previous leader directed us to join new leader {}'.format(newleader)})
try:
if await connect_to_leader(None, get_myname(), newleader):
return
handled = True
except Exception:
log.log({'error': 'Unknown error attempting to connect to {}, check trace log'.format(newleader), 'subsystem': 'collective'})
cfm.logException()
log.log({'info': 'Current leader ({0}) has disappeared, restarting '
'collective membership'.format(leader), 'subsystem': 'collective'})
# The leader has folded, time to startup again...
follower = None
await cfm.stop_following()
currentleader = None
if retrythread is None: # start a recovery
retrythread = tasks.spawn_task_after(
random.random(), start_collective)
if not handled:
log.log({'info': 'Current leader ({0}) has disappeared, restarting '
'collective membership'.format(leader), 'subsystem': 'collective'})
# The leader has folded, time to startup again...
follower = None
await cfm.stop_following()
currentleader = None
if retrythread is None: # start a recovery
retrythread = tasks.spawn_task_after(
random.random(), start_collective)
async def _create_tls_connection(host, port):
cloop = asyncio.get_running_loop()
confluent_server/confluent/config/configmanager.py
+7 -4
View File
@@ -782,6 +782,9 @@ async def relay_slaved_requests(name, listener):
msg = await lh.get_next_msg()
except Exception:
msg = None
except Exception:
# unexpected loss or misbehavior of the follower ends the relay
logException()
finally:
try:
listener.close()
@@ -798,10 +801,10 @@ async def relay_slaved_requests(name, listener):
*[_push_rpc(cfgstreams[s]['stream'], payload) for s in cfgstreams])
if membership_callback:
membership_callback()
if not cfgstreams and not cfgleader: # last one out, set cfgleader to boolean to mark dead collective
await stop_following(True)
return False
return True
if not cfgstreams and not cfgleader: # last one out, set cfgleader to boolean to mark dead collective
await stop_following(True)
return False
return True
lastheartbeat = None
async def check_leader():
confluent_server/confluent/log.py
+1 -1
View File
@@ -396,7 +396,7 @@ class TimedAndSizeRotatingFileHandler(BaseRotatingHandler):
prefix = baseName + "."
if dirName not in self.dirContents or self.dirContents[dirName][1] < time.time():
self.dirContents[dirName] = (os.listdir(dirName), time.time() + 5)
matchexp = re.compile(f'^{prefix}\.\d+$')
matchexp = re.compile(rf'^{prefix}\.\d+$')
fileNames = [f for f in self.dirContents[dirName][0]
if matchexp.match(f)]
plen = len(prefix)
confluent_server/confluent/plugins/hardwaremanagement/enos.py
+9 -9
View File
@@ -152,12 +152,12 @@ def gather_data(configmanager, creds, node):
nssh = enos_login(node=node, configmanager=configmanager, creds=creds)
switch_lines = enos_version(ssh=nssh)
switch_data = {}
sysinfo = {"Product name": {"regex": ".*RackSwitch (\w+)"},
"Serial Number": {"regex": "ESN\s*\w*\s*: ([\w-]+)"},
"Board Serial Number": {"regex": "Switch Serial No: (\w+)"},
"Model": {"regex": "MTM\s*\w*\s*: ([\w-]+)"},
"FRU Number": {"regex": "Hardware Part\s*\w*\s*: (\w+)"},
"Airflow": {"regex": "System Fan Airflow\s*\w*\s*: ([\w-]+)"},
sysinfo = {"Product name": {"regex": r".*RackSwitch (\w+)"},
"Serial Number": {"regex": r"ESN\s*\w*\s*: ([\w-]+)"},
"Board Serial Number": {"regex": r"Switch Serial No: (\w+)"},
"Model": {"regex": r"MTM\s*\w*\s*: ([\w-]+)"},
"FRU Number": {"regex": r"Hardware Part\s*\w*\s*: (\w+)"},
"Airflow": {"regex": r"System Fan Airflow\s*\w*\s*: ([\w-]+)"},
}
invinfo = {
@@ -233,7 +233,7 @@ def gather_data(configmanager, creds, node):
sysfw = {"Software Version": "Unknown", "Boot kernel": "Unknown"}
for line in switch_lines:
for key in sysfw.keys():
regex = f"{key}\s*\w*\s* ([0-9.]+)"
regex = rf"{key}\s*\w*\s* ([0-9.]+)"
match = re.match(re.compile(regex), line)
if match:
sysfw[key] = match.group(1)
@@ -250,7 +250,7 @@ def gather_psus(data):
# others are:
# Internal Power Supply: On
if "Power Supply" in line:
match = re.match(re.compile("Power Supply (\d)+.*"), line)
match = re.match(re.compile(r"Power Supply (\d)+.*"), line)
if match:
psu = match.group(1)
if psu not in psus:
@@ -280,7 +280,7 @@ def gather_fans(data):
for line in data:
# look for presence of fans
if "Fan" in line:
match = re.match(re.compile("Fan (\d)+.*"), line)
match = re.match(re.compile(r"Fan (\d)+.*"), line)
if match:
fan = match.group(1)
if match:
confluent_server/confluent_server.spec.tmpl
+1 -4
View File
@@ -68,10 +68,7 @@ chown confluent:confluent /etc/confluent /var/lib/confluent /var/log/confluent /
sysctl -p /usr/lib/sysctl.d/confluent.conf >& /dev/null
NEEDCHOWN=0
NEEDSTART=0
find /etc/confluent -uid 0 | grep -E '.*' > /dev/null && NEEDCHOWN=1
find /var/log/confluent -uid 0 | grep -E '.*' > /dev/null && NEEDCHOWN=1
find /var/run/confluent -uid 0 | grep -E '.*' > /dev/null && NEEDCHOWN=1
find /var/cache/confluent -uid 0 | grep -E '.*' > /dev/null && NEEDCHOWN=1
[ -n "$(find /etc/confluent /var/log/confluent /var/cache/confluent -uid 0 -print -quit 2>/dev/null)" ] && NEEDCHOWN=1
if [ $NEEDCHOWN = 1 ]; then
if systemctl is-active confluent > /dev/null; then
NEEDSTART=1
confluent_server/sysctl/confluent.conf
+1 -1
View File
@@ -1,2 +1,2 @@
# Increase available receive buffers for discovery scans
net.core.rmem_max = 2097152
net.core.rmem_max = 4194304