Test: don't verify ES SSL cert. Remap ES test data (#247)

This commit will have the integration testing app skip verification of
the SSL certificate Elasticsearch server uses (in case of indexing into
an HTTPS URL). This simplifies loading data into nodes with self-signed
or invalid certs.

The commit also now deletes the "old" templates and pipelines before
reindexing.

And finally, it generates indices with changed mappings for
Elasticsearch own test indices to match the field types used in ES
tests. (The "original" indices are still kept since they CSV
reconstruction tests are easier with them.)

Author: bpintea

test/integration/data.py

@@ -5,6 +5,7 @@
 #
 
 import requests
+import urllib3
 import io
 import csv
 import json
@@ -205,6 +206,81 @@
 		]
 	}
 
+EMP_TEST_MAPPING =\
+	{
+		"mappings" : {
+			"properties" : {
+				"birth_date" : {
+					"type" : "date"
+				},
+				"emp_no" : {
+					"type" : "integer"
+				},
+				"first_name" : {
+					"type" : "text",
+					"fields" : {
+						"keyword" : {
+							"type" : "keyword",
+							"ignore_above" : 256
+						}
+					}
+				},
+				"gender" : {
+					"type" : "keyword"
+				},
+				"hire_date" : {
+					"type" : "date"
+				},
+				"languages" : {
+					"type" : "integer"
+				},
+				"last_name" : {
+					"type" : "text",
+					"fields" : {
+						"keyword" : {
+							"type" : "keyword",
+							"ignore_above" : 256
+						}
+					}
+				},
+				"salary" : {
+					"type" : "long"
+				}
+			}
+		}
+	}
+
+LIB_TEST_MAPPING =\
+	{
+		"mappings": {
+			"properties": {
+				"author": {
+					"type": "text",
+					"fields": {
+						"keyword": {
+							"type": "keyword",
+							"ignore_above": 256
+						}
+					}
+				},
+				"name": {
+					"type": "text",
+					"fields": {
+						"keyword": {
+							"type": "keyword",
+							"ignore_above": 256
+						}
+					}
+				},
+				"page_count": {
+					"type": "integer"
+				},
+				"release_date": {
+					"type": "date"
+				}
+			}
+		}
+	}
 
 ES_DATASET_BASE_URL = "https://raw.githubusercontent.com/elastic/elasticsearch/eda31b0ac00c952a52885902be59ac429b0ca81a/x-pack/plugin/sql/qa/src/main/resources/"
 
@@ -266,8 +342,10 @@ class TestData(object):
 	BATTERS_INDEX = "batters"
 	LIBRARY_FILE = "library.csv"
 	LIBRARY_INDEX = "library"
+	LIB_TEST_INDEX = "test_lib"
 	EMPLOYEES_FILE = "employees.csv"
 	EMPLOYEES_INDEX = "employees"
+	EMP_TEST_INDEX = "test_emp"
 	PROTO_CASE_FILE = "SqlProtocolTestCase.java"
 
 
@@ -292,6 +370,8 @@ class TestData(object):
 		MODE_INDEX: "indexed"
 	}
 
+	_session = None
+
 	def __init__(self, es, mode=MODE_INDEX, offline_dir=None):
 		self._csv_md5 = {}
 		self._csv_header = {}
@@ -301,6 +381,10 @@ def __init__(self, es, mode=MODE_INDEX, offline_dir=None):
 		self._offline_dir = offline_dir
 		self._mode = mode
 
+		self._req = requests.Session()
+		self._req.verify = False
+		urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
 	def _csv_to_json_docs(self, csv_text):
 		stream = io.StringIO(csv_text)
 		reader = csv.reader(stream, delimiter=',', quotechar='"')
@@ -374,24 +458,11 @@ def _get_csv_as_ndjson(self, base_url, csv_name, index_name):
 		else:
 			assert(base_url.endswith("/"))
 			url = base_url + csv_name
-			req = requests.get(url, timeout = Elasticsearch.REQ_TIMEOUT)
+			req = self._req.get(url, timeout = Elasticsearch.REQ_TIMEOUT)
 			if req.status_code != 200:
 				raise Exception("failed to fetch %s with code %s" % (url, req.status_code))
 			return self._csv_as_ndjson(req.text, req.encoding, index_name)
 
-
-	def _prepare_tableau_load(self, file_name, index_name, index_template):
-		ndjson = self._get_csv_as_ndjson(TABLEAU_DATASET_BASE_URL, file_name, index_name)
-
-		if self.MODE_NOINDEX < self._mode:
-			with requests.put("%s/_template/%s_template" % (self._es.base_url(), index_name),
-					json=index_template, auth=self._es.credentials()) as req:
-				if req.status_code != 200:
-					raise Exception("PUT %s template failed with code: %s (content: %s)" % (index_name,
-						req.status_code, req.text))
-
-		return ndjson
-
 	def _post_ndjson(self, ndjsons, index_name, pipeline_name=None):
 		print("Indexing data for index '%s'." % index_name)
 		url = "%s/%s/_bulk" % (self._es.base_url(), index_name)
@@ -400,7 +471,7 @@ def _post_ndjson(self, ndjsons, index_name, pipeline_name=None):
 		if type(ndjsons) is not list:
 			ndjsons = [ndjsons]
 		for n in ndjsons:
-			with requests.post(url, data=n, headers = {"Content-Type": "application/x-ndjson"},
+			with self._req.post(url, data=n, headers = {"Content-Type": "application/x-ndjson"},
 					auth=self._es.credentials()) as req:
 				if req.status_code not in [200, 201]:
 					raise Exception("bulk POST to %s failed with code: %s (content: %s)" % (index_name,
@@ -415,7 +486,7 @@ def _wait_for_results(self, index_name):
 		waiting_since = time.time()
 		while hits < MIN_INDEXED_DOCS:
 			url = "%s/%s/_search" % (self._es.base_url(), index_name)
-			req = requests.get(url, timeout = Elasticsearch.REQ_TIMEOUT, auth=self._es.credentials())
+			req = self._req.get(url, timeout = Elasticsearch.REQ_TIMEOUT, auth=self._es.credentials())
 			if req.status_code != 200:
 				raise Exception("failed to _search %s: code: %s, body: %s" % (index_name, req.status_code, req.text))
 			answer = json.loads(req.text)
@@ -424,32 +495,48 @@ def _wait_for_results(self, index_name):
 			if Elasticsearch.REQ_TIMEOUT < time.time() - waiting_since:
 				raise Exception("index '%s' has less than %s documents indexed" % (index_name, MIN_INDEXED_DOCS))
 
-	def _delete_if_needed(self, index_name):
+	def _del_resource(self, url):
+		with self._req.delete(url, timeout = Elasticsearch.REQ_TIMEOUT, auth=self._es.credentials()) as req:
+			if req.status_code != 200 and req.status_code != 404:
+				raise Exception("Deleting %s failed; code=%s, body: %s." % (url, req.status_code, req.text))
+
+	def _delete_if_needed(self, index_name, template=False, pipeline=False):
 		if self._mode != self.MODE_REINDEX:
 			return
-		print("Deleting any old index '%s'." % index_name);
+		print("Deleting any old: index '%s'." % index_name);
 
 		url = "%s/%s" % (self._es.base_url(), index_name)
-		with requests.delete(url, timeout = Elasticsearch.REQ_TIMEOUT, auth=self._es.credentials()) as req:
-			if req.status_code != 200 and req.status_code != 404:
-				raise Exception("Deleting index %s failed; code=%s, body: %s." %
-						(index_name, req.status_code, req.text))
+		self._del_resource(url)
+
+		if template:
+			url = "%s/_template/%s" % (self._es.base_url(), index_name)
+			self._del_resource(url)
+
+		if pipeline:
+			url = "%s/_ingest/pipeline/%s" % (self._es.base_url(), index_name)
+			self._del_resource(url)
 
 	def _load_tableau_sample(self, file_name, index_name, template, pipeline=None):
-		ndjsons = self._prepare_tableau_load(file_name, index_name, template)
+		if self._mode <= self.MODE_NOINDEX:
+			return
+		self._delete_if_needed(index_name, True, pipeline is not None)
 
-		if self.MODE_NOINDEX < self._mode:
-			self._delete_if_needed(index_name)
+		with self._req.put("%s/_template/%s" % (self._es.base_url(), index_name),
+				json=template, auth=self._es.credentials()) as req:
+			if req.status_code != 200:
+				raise Exception("PUT %s template failed with code: %s (content: %s)" % (index_name,
+					req.status_code, req.text))
 
-			if pipeline:
-				with requests.put("%s/_ingest/pipeline/parse_%s" % (self._es.base_url(), index_name),
-						json=pipeline, auth=self._es.credentials()) as req:
-					if req.status_code != 200:
-						raise Exception("PUT %s pipeline failed with code: %s (content: %s) " % (index_name,
-							req.status_code, req.text))
+		if pipeline:
+			with self._req.put("%s/_ingest/pipeline/%s" % (self._es.base_url(), index_name),
+					json=pipeline, auth=self._es.credentials()) as req:
+				if req.status_code != 200:
+					raise Exception("PUT %s pipeline failed with code: %s (content: %s) " % (index_name,
+						req.status_code, req.text))
 
-			self._post_ndjson(ndjsons, index_name, ("parse_" + index_name) if pipeline else None)
-			self._wait_for_results(index_name)
+		ndjsons = self._get_csv_as_ndjson(TABLEAU_DATASET_BASE_URL, file_name, index_name)
+		self._post_ndjson(ndjsons, index_name, index_name if pipeline else None)
+		self._wait_for_results(index_name)
 
 	def _load_elastic_sample(self, file_name, index_name):
 		ndjson = self._get_csv_as_ndjson(ES_DATASET_BASE_URL, file_name, index_name)
@@ -458,6 +545,25 @@ def _load_elastic_sample(self, file_name, index_name):
 			self._post_ndjson(ndjson, index_name)
 			self._wait_for_results(index_name)
 
+	def _derive_with_mapping(self, src_index, dst_index, mapping_json):
+		if self._mode < self.MODE_REINDEX:
+			return
+		print("Reindexing '%s' into '%s'." % (src_index, dst_index))
+		self._delete_if_needed(dst_index)
+
+		with self._req.put("%s/%s" % (self._es.base_url(), dst_index),
+				json=mapping_json, auth=self._es.credentials()) as req:
+			if req.status_code != 200:
+				raise Exception("PUT %s mapping failed with code: %s (content: %s) " % (dst_index,
+					req.status_code, req.text))
+
+		reindex_json = {"source": {"index": src_index}, "dest": {"index": dst_index}}
+		with self._req.post("%s/_reindex?wait_for_completion=true" % self._es.base_url(),
+				json=reindex_json, auth=self._es.credentials()) as req:
+			if req.status_code != 200:
+				raise Exception("POST reindexing into %s failed with code: %s (content: %s) " % (dst_index,
+					req.status_code, req.text))
+
 	def _get_kibana_file(self, sample_name, is_mapping=True):
 		print("Fetching JS sample data for index '%s'." % sample_name)
 		file_name = "field_mappings.js" if is_mapping else "%s.json.gz" % sample_name
@@ -468,7 +574,7 @@ def _get_kibana_file(self, sample_name, is_mapping=True):
 		else:
 			url = KIBANA_SAMPLES_BASE_URL + "/" + sample_name + "/"
 			url += file_name
-			req = requests.get(url, timeout = Elasticsearch.REQ_TIMEOUT)
+			req = self._req.get(url, timeout = Elasticsearch.REQ_TIMEOUT)
 			if req.status_code != 200:
 				raise Exception("failed to GET URL %s for index %s with: code: %s, body: %s" %
 						(url, sample_name, req.status_code, req.text))
@@ -492,8 +598,8 @@ def _put_sample_template(self, sample_name, index_name):
 		# turn it to JSON (to deal with trailing commas past last member on a level
 		mapping = eval(mapping)
 		# PUT the built template
-		url = "%s/_template/%s_template" % (self._es.base_url(), index_name)
-		with requests.put(url, json=mapping, auth=self._es.credentials(), timeout=Elasticsearch.REQ_TIMEOUT) as req:
+		url = "%s/_template/%s" % (self._es.base_url(), index_name)
+		with self._req.put(url, json=mapping, auth=self._es.credentials(), timeout=Elasticsearch.REQ_TIMEOUT) as req:
 			if req.status_code != 200:
 				raise Exception("PUT %s template failed with code: %s (content: %s)" % (index_name,
 						req.status_code, req.text))
@@ -522,7 +628,7 @@ def _load_proto_tests(self):
 				case_src = f.read()
 		else:
 			url = ES_PROTO_CASE_BASE_URL + "/" + self.PROTO_CASE_FILE
-			req = requests.get(url, timeout=Elasticsearch.REQ_TIMEOUT)
+			req = self._req.get(url, timeout=Elasticsearch.REQ_TIMEOUT)
 			if req.status_code != 200:
 				raise Exception("failed to fetch %s with code %s" % (url, req.status_code))
 			case_src = req.text
@@ -550,7 +656,9 @@ def load(self):
 			self._load_tableau_sample(self.BATTERS_FILE, self.BATTERS_INDEX, BATTERS_TEMPLATE, BATTERS_PIPELINE)
 
 			self._load_elastic_sample(self.LIBRARY_FILE, self.LIBRARY_INDEX)
+			self._derive_with_mapping(self.LIBRARY_INDEX, self.LIB_TEST_INDEX, LIB_TEST_MAPPING)
 			self._load_elastic_sample(self.EMPLOYEES_FILE, self.EMPLOYEES_INDEX)
+			self._derive_with_mapping(self.EMPLOYEES_INDEX, self.EMP_TEST_INDEX, EMP_TEST_MAPPING)
 
 			self._load_kibana_sample(self.ECOMMERCE_INDEX)
 			self._load_kibana_sample(self.FLIGHTS_INDEX)

test/integration/elasticsearch.py

@@ -5,6 +5,7 @@
 #
 
 import requests
+import urllib3
 import psutil
 
 import sys
@@ -43,6 +44,8 @@ class Elasticsearch(object):
 	_credentials = None
 	_base_url = None
 
+	_req = None
+
 	def __init__(self, offline_dir=None, url=None):
 		self._offline_dir = offline_dir
 		if not url:
@@ -57,6 +60,10 @@ def __init__(self, offline_dir=None, url=None):
 		print("Using Elasticsearch instance at %s, credentials (%s, %s)" % (self._base_url, self._credentials[0],
 			"*" * len(self._credentials[1])))
 
+		self._req = requests.session()
+		self._req.verify = False
+		urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
 	@staticmethod
 	def elasticsearch_distro_filename(version):
 		return "%s-%s%s.%s" % (ES_PROJECT, version, ES_ARCH, PACKAGING)
@@ -68,13 +75,13 @@ def base_url(self):
 		return self._base_url
 
 	def _latest_build(self, version):
-		req = requests.get(ARTIF_URL, timeout=self.REQ_TIMEOUT)
+		req = self._req.get(ARTIF_URL, timeout=self.REQ_TIMEOUT)
 		vers = req.json()["versions"]
 		if version not in vers:
 			raise Exception("version %s not found; available: %s" % (version, vers))
 
 		builds_url = ARTIF_URL + "/" + version + "/builds"
-		req = requests.get(builds_url, timeout=self.REQ_TIMEOUT)
+		req = self._req.get(builds_url, timeout=self.REQ_TIMEOUT)
 		builds = req.json()["builds"]
 
 		# file name to download
@@ -86,7 +93,7 @@ def _latest_build(self, version):
 		# requests.async: no Windows support
 		for build in builds:
 			build_url = builds_url + "/" + build
-			req = requests.get(build_url, timeout=self.REQ_TIMEOUT)
+			req = self._req.get(build_url, timeout=self.REQ_TIMEOUT)
 			build_json = req.json()["build"]
 			dt_et = datetime.strptime(build_json["end_time"], "%a, %d %b %Y %X %Z")
 			if latest_et < dt_et:
@@ -103,7 +110,7 @@ def _latest_build(self, version):
 	def _download_build(self, url, dest_dir):
 		print("- downloading %s: " % url, end='')
 		size = 0
-		req = requests.get(url, stream=True, timeout=self.REQ_TIMEOUT)
+		req = self._req.get(url, stream=True, timeout=self.REQ_TIMEOUT)
 		dest_file = os.path.join(dest_dir, req.url[req.url.rfind('/')+1 : ])
 		with open(dest_file, "wb") as f:
 			for chunk in req.iter_content(chunk_size = 32 * K1):
@@ -217,12 +224,12 @@ def _enable_xpack(self, es_dir):
 		# setup passwords to random generated ones first...
 		pwd = self._gen_passwords(es_dir)
 		# ...then change passwords, easier to restart with failed tests
-		req = requests.post("%s/_security/user/_password" % self._base_url, auth=(self._credentials[0], pwd),
+		req = self._req.post("%s/_security/user/_password" % self._base_url, auth=(self._credentials[0], pwd),
 				json={"password": self._credentials[1]})
 		if req.status_code != 200:
 			raise Exception("attempt to change elastic's password failed with code %s" % req.status_code)
 		# kibana too (debug convenience)
-		req = requests.post("%s/_security/user/kibana/_password" % self._base_url, auth=self._credentials,
+		req = self._req.post("%s/_security/user/kibana/_password" % self._base_url, auth=self._credentials,
 				json={"password": self._credentials[1]})
 		if req.status_code != 200:
 			print("ERROR: kibana user password change failed with code: %s" % req.status_code)
@@ -231,7 +238,7 @@ def _enable_xpack(self, es_dir):
 		url = "%s/_license/start_trial?acknowledge=true" % self._base_url
 		failures = 0
 		while True:
-			req = requests.post(url, auth=self._credentials, timeout=self.REQ_TIMEOUT)
+			req = self._req.post(url, auth=self._credentials, timeout=self.REQ_TIMEOUT)
 			if req.status_code == 200:
 				# TODO: check content?
 				break
@@ -284,7 +291,7 @@ def reset(self, es_dir, reset_conf=False):
 
 	def cluster_name(self, fail_on_non200=True):
 		try:
-			resp = requests.get(self._base_url, auth=self._credentials, timeout=self.REQ_TIMEOUT)
+			resp = self._req.get(self._base_url, auth=self._credentials, timeout=self.REQ_TIMEOUT)
 		except (requests.Timeout, requests.ConnectionError):
 			return None
 		if resp.status_code != 200: