This is an automated email from the ASF dual-hosted git repository.
sebb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-ponymail-foal.git
The following commit(s) were added to refs/heads/master by this push:
new 4c3c12c Drop use of dbname index prefix
4c3c12c is described below
commit 4c3c12c01c4118ea40b9ccc688cc4046a088e76e
Author: Sebb <[email protected]>
AuthorDate: Tue Sep 28 14:16:40 2021 +0100
Drop use of dbname index prefix
This fixes #107
---
tools/import-mbox.py | 22 ++++++++++------------
tools/plugins/elastic.py | 31 ++++++++++++++-----------------
2 files changed, 24 insertions(+), 29 deletions(-)
diff --git a/tools/import-mbox.py b/tools/import-mbox.py
index b432486..93906ae 100755
--- a/tools/import-mbox.py
+++ b/tools/import-mbox.py
@@ -84,7 +84,7 @@ noMboxo = False # Don't skip MBoxo patch
rootURL = ""
-def bulk_insert(name, json, xes, dtype, wc="quorum"):
+def bulk_insert(name, json, xes, dbindex, wc="quorum"):
sys.stderr.flush()
@@ -92,13 +92,13 @@ def bulk_insert(name, json, xes, dtype, wc="quorum"):
for entry in json:
js = entry
document_id = js["mid"]
- if dtype == "source":
+ if dbindex == xes.db_source:
del js["mid"]
js_arr.append(
{
"_op_type": "index",
"_consistency": wc,
- "_index": dbname + "-" + dtype,
+ "_index": dbindex,
"_id": document_id,
"doc": js,
"_source": js,
@@ -106,9 +106,9 @@ def bulk_insert(name, json, xes, dtype, wc="quorum"):
)
try:
xes.bulk(js_arr, ignore=404)
- # print("%s: Inserted %u entries into %s" % (name,
len(js_arr),dtype))
+ # print("%s: Inserted %u entries into %s" % (name,
len(js_arr),dbindex))
except Exception as err:
- print("%s: Warning: Could not bulk insert: %s into %s" % (name, err,
dtype))
+ print("%s: Warning: Could not bulk insert: %s into %s" % (name, err,
dbindex))
class DownloadThread(Thread):
def assign(self, url):
@@ -385,10 +385,10 @@ class SlurpThread(Thread):
body={"source": contents[key]},
)
if len(ja) >= 40 and not args.dry:
- bulk_insert(self.name, ja, es, "mbox")
+ bulk_insert(self.name, ja, es, es.db_mbox)
ja = []
- bulk_insert(self.name, jas, es, "source")
+ bulk_insert(self.name, jas, es, es.db_source)
jas = []
else:
self.printid(
@@ -415,11 +415,11 @@ class SlurpThread(Thread):
goodies += count
baddies += bad
if len(ja) > 0 and not args.dry:
- bulk_insert(self.name, ja, es, "mbox")
+ bulk_insert(self.name, ja, es, es.db_mbox)
ja = []
if len(jas) > 0 and not args.dry:
- bulk_insert(self.name, jas, es, "source")
+ bulk_insert(self.name, jas, es, es.db_source)
jas = []
self.printid("Done, %u elements left to slurp" % len(lists))
@@ -625,11 +625,9 @@ if args.dry:
else:
# Fetch config and set up ES
es = Elastic()
- # We need the index name for bulk actions
- dbname = es.getdbname()
# No point continuing if the index does not exist
- print("Checking that the database index %s exists ... " % dbname)
+ print("Checking that the database index %s exists ... " % es.db_mbox)
# Need to check the index before starting bulk operations
try:
diff --git a/tools/plugins/elastic.py b/tools/plugins/elastic.py
index bd8d9ec..8dca607 100755
--- a/tools/plugins/elastic.py
+++ b/tools/plugins/elastic.py
@@ -51,15 +51,15 @@ class Elastic:
config = ponymailconfig.PonymailConfig()
# Set default names for all indices we use
- self.dbname = config.get('elasticsearch', 'dbname',
fallback='ponymail')
- self.db_mbox = self.dbname + '-mbox'
- self.db_source = self.dbname + '-source'
- self.db_account = self.dbname + '-account'
- self.db_attachment = self.dbname + '-attachment'
- self.db_session = self.dbname + '-session'
- self.db_notification = self.dbname + '-notification'
- self.db_mailinglist = self.dbname + '-mailinglist'
- self.db_auditlog = self.dbname + '-auditlog'
+ dbname = config.get('elasticsearch', 'dbname', fallback='ponymail')
+ self.db_mbox = dbname + '-mbox'
+ self.db_source = dbname + '-source'
+ self.db_account = dbname + '-account'
+ self.db_attachment = dbname + '-attachment'
+ self.db_session = dbname + '-session'
+ self.db_notification = dbname + '-notification'
+ self.db_mailinglist = dbname + '-mailinglist'
+ self.db_auditlog = dbname + '-auditlog'
self.db_version = 0
dburl = config.get('elasticsearch', 'dburl', fallback=None)
@@ -131,11 +131,8 @@ class Elastic:
def engineMajor(self):
return int(self.engineVersion().split(".")[0])
- def getdbname(self):
- return self.dbname
-
def search(self, **kwargs):
- return self.es.search(index=self.dbname, **kwargs)
+ return self.es.search(**kwargs)
def index(self, **kwargs):
kwargs["wait_for_active_shards"] = self.wait_for_active_shards
@@ -149,11 +146,11 @@ class Elastic:
return self.es.info(**kwargs)
def update(self, **kwargs):
- return self.es.update(index=self.dbname, **kwargs)
+ return self.es.update(**kwargs)
def scan(self, scroll="3m", size=100, **kwargs):
return self.es.search(
- index=self.dbname, search_type="scan", size=size, scroll=scroll,
**kwargs
+ search_type="scan", size=size, scroll=scroll, **kwargs
)
def scan_and_scroll(self, scroll="3m", size=100, **kwargs):
@@ -162,7 +159,7 @@ class Elastic:
incorporates es.scroll for continuous iteration, and thus the
scroll() does NOT need to be called at all by the calling
process. """
- results = self.es.search(index=self.dbname, size=size, scroll=scroll,
**kwargs)
+ results = self.es.search(size=size, scroll=scroll, **kwargs)
if results["hits"].get("hits", []): # Might not be there in 2.x?
yield results
@@ -176,7 +173,7 @@ class Elastic:
yield results
def get(self, **kwargs):
- return self.es.get(index=self.dbname, **kwargs)
+ return self.es.get(**kwargs)
def scroll(self, **kwargs):
return self.es.scroll(**kwargs)
