#
|
# The contents of this file are subject to the terms of the Common Development and
|
# Distribution License (the License). You may not use this file except in compliance with the
|
# License.
|
#
|
# You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
|
# specific language governing permission and limitations under the License.
|
#
|
# When distributing Covered Software, include this CDDL Header Notice in each file and include
|
# the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
|
# Header, with the fields enclosed by brackets [] replaced by your own identifying
|
# information: "Portions copyright [year] [name of copyright owner]".
|
#
|
# Copyright 2026 3A Systems, LLC.
|
|
"""Finds the changes a replication server's changelog lacks (#1090, #1195).
|
|
A directory server numbers the changes it makes one after the other: CSNGenerator.newCSN()
|
adds one to the sequence number of the CSN for every change, whatever its time. So the
|
changes of one replica in a changelog have no gap in their sequence numbers unless a change
|
was never published - an operation that failed after it had its CSN, or one that was lost on
|
its way to the replication servers. During the load of the convergence run no operation fails
|
(load.sh checks that), so a gap there is a lost change: a change the data of the other
|
replicas misses, while their ServerState and missing-changes say nothing, since both only
|
know the newest CSN.
|
|
Usage: changelog-gaps.py <base DN> <from ms> <to ms> [<changelogDb directory>]
|
|
Without a directory it reads the changelogDb as a tar stream on its standard input, as
|
"docker cp <container>:/opt/opendj/data/changelogDb -" writes it. It reports, for every
|
replica of the domain of <base DN>, the changes whose CSN time is between <from ms> and
|
<to ms>, and the sequence numbers missing between them. Exit code 1: a change is missing;
|
2: the changelog cannot be read (no such domain, or it is encrypted).
|
|
What it cannot tell. A change lost at either end of a replica's changes in the window - its
|
last ones, or its first ones when no earlier change of the replica is in the changelog - leaves
|
no gap between two changes that are there; the data of the other replicas shows such a change
|
unless a later change of another server overwrites it. And not every sequence number that no
|
record holds is a lost change: a directory server that goes offline (a restart, its domain
|
disabled) numbers its ReplicaOfflineMsg like a change, and the changelog keeps only the newest
|
of them, in the offline.state of the replica, not in a log file. The convergence run restarts
|
no server after the topology is up; a run that did would read each restart in the window as a
|
lost change.
|
|
The file format is the one of FileReplicaDB: every log file is cut into blocks of 256 bytes,
|
and every block after the first starts with a 4-byte offset (BlockLogWriter); in between, a
|
record is its length in 4 bytes and the bytes of its UpdateMsg, whose header is the message
|
type, the protocol version and the CSN as 28 hexadecimal characters (LDAPUpdateMsg).
|
"""
|
|
import os
|
import re
|
import sys
|
import tarfile
|
|
BLOCK_SIZE = 256
|
SIZE_OF_BLOCK_OFFSET = 4
|
# ReplicationMsg: MSG_TYPE_MODIFY, ADD, DELETE, MODIFYDN of the protocol versions after 1
|
UPDATE_TYPES = {22: "modify", 23: "add", 24: "delete", 25: "modifydn"}
|
DISK_ENCODING = 0xFF
|
CSN = re.compile(rb"[0-9a-f]{28}")
|
LOG_FILE = re.compile(r"(?:^|/)(\d+)\.dom/(\d+)\.server/[^/.][^/]*\.log$")
|
|
|
def records(data):
|
"""The UpdateMsg bytes of the records of one log file."""
|
stream = bytearray(data[:BLOCK_SIZE])
|
for start in range(BLOCK_SIZE, len(data), BLOCK_SIZE):
|
stream += data[start + SIZE_OF_BLOCK_OFFSET:start + BLOCK_SIZE]
|
pos = 0
|
while pos + 4 <= len(stream):
|
length = int.from_bytes(stream[pos:pos + 4], "big")
|
if length <= 0 or pos + 4 + length > len(stream):
|
raise ValueError("record of %d bytes at %d of %d" % (length, pos, len(stream)))
|
yield bytes(stream[pos + 4:pos + 4 + length])
|
pos += 4 + length
|
|
|
def read_changelog(entries):
|
"""(domains, changes): domain id by base DN, and for each (domain id, server id) a
|
list of (sequence number, time in ms, type)."""
|
domains = {}
|
changes = {}
|
for name, data in entries:
|
if os.path.basename(name) == "domains.state":
|
for line in data.decode("utf-8").splitlines():
|
if ":" in line:
|
domain_id, dn = line.split(":", 1)
|
domains[dn.strip().lower()] = domain_id
|
continue
|
match = LOG_FILE.search(name)
|
if not match:
|
continue
|
found = changes.setdefault((match.group(1), int(match.group(2))), [])
|
for message in records(data):
|
if message[0] == DISK_ENCODING:
|
raise EncryptedChangelog(name)
|
if message[0] not in UPDATE_TYPES or not CSN.fullmatch(message[2:30]):
|
continue
|
csn = message[2:30].decode("ascii")
|
found.append((int(csn[20:28], 16), int(csn[0:16], 16), UPDATE_TYPES[message[0]]))
|
return domains, changes
|
|
|
class EncryptedChangelog(Exception):
|
pass
|
|
|
def from_directory(root):
|
for directory, _, files in os.walk(root):
|
for file in files:
|
path = os.path.join(directory, file)
|
with open(path, "rb") as f:
|
yield os.path.relpath(path, root), f.read()
|
|
|
def from_tar(stream):
|
with tarfile.open(fileobj=stream, mode="r|") as tar:
|
for member in tar:
|
if member.isfile():
|
yield member.name, tar.extractfile(member).read()
|
|
|
def main(argv):
|
if len(argv) not in (4, 5):
|
sys.stderr.write(__doc__)
|
return 2
|
base_dn, start, end = argv[1].lower(), int(argv[2]), int(argv[3])
|
entries = from_directory(argv[4]) if len(argv) == 5 else from_tar(sys.stdin.buffer)
|
try:
|
domains, changes = read_changelog(entries)
|
except EncryptedChangelog as e:
|
print("the changelog is encrypted (%s): its changes cannot be read" % e)
|
return 2
|
except (tarfile.TarError, ValueError, OSError) as e:
|
print("the changelog cannot be read: %s" % e)
|
return 2
|
if base_dn not in domains:
|
print("no domain %s in the changelog, which holds %s" % (base_dn, sorted(domains)))
|
return 2
|
lost = False
|
for (domain_id, server_id), found in sorted(changes.items()):
|
if domain_id != domains[base_dn]:
|
continue
|
load = sorted((seqnum, time, kind) for seqnum, time, kind in found if start <= time <= end)
|
if not load:
|
continue
|
seqnums = [seqnum for seqnum, _, _ in load]
|
missing = []
|
for previous, following in zip(load, load[1:]):
|
missing += [(seqnum, previous, following) for seqnum in range(previous[0] + 1, following[0])]
|
print("replica %d: %d changes in the load, sequence numbers %d to %d, %d missing"
|
% (server_id, len(load), seqnums[0], seqnums[-1], len(missing)))
|
for seqnum, previous, following in missing[:20]:
|
print(" missing %d, between the %s %d (%d) and the %s %d (%d)"
|
% (seqnum, previous[2], previous[0], previous[1], following[2], following[0], following[1]))
|
if len(missing) > 20:
|
print(" and %d more" % (len(missing) - 20))
|
lost = lost or bool(missing)
|
return 1 if lost else 0
|
|
|
if __name__ == "__main__":
|
sys.exit(main(sys.argv))
|
