Changeset: fd01b17e9fea for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=fd01b17e9fea
Modified Files:
clients/iotclient/src/Flask/app.py
clients/iotclient/src/Settings/filesystem.py
clients/iotclient/src/Settings/mapiconnection.py
clients/iotclient/src/Streams/streams.py
clients/iotclient/src/main.py
Branch: iot
Log Message:
Added option to add hostname as a stream column.
diffs (truncated from 306 to 300 lines):
diff --git a/clients/iotclient/src/Flask/app.py
b/clients/iotclient/src/Flask/app.py
--- a/clients/iotclient/src/Flask/app.py
+++ b/clients/iotclient/src/Flask/app.py
@@ -1,7 +1,7 @@
from flask import Flask
from flask_restful import Api
-from restresources import StreamInput, StreamsInfo, StreamsHandling # ,
ServerHandler
+from restresources import StreamInput, StreamsInfo, StreamsHandling
def start_flask_iot_app(host, port):
@@ -21,5 +21,4 @@ def start_flask_admin_app(host, port):
admin_api.add_resource(StreamsInfo, '/streams')
admin_api.add_resource(StreamsHandling, '/context')
- # admin_api.add_resource(ServerHandler, '/server')
admin_app.run(host=host, port=port, threaded=True)
diff --git a/clients/iotclient/src/Settings/filesystem.py
b/clients/iotclient/src/Settings/filesystem.py
--- a/clients/iotclient/src/Settings/filesystem.py
+++ b/clients/iotclient/src/Settings/filesystem.py
@@ -5,22 +5,24 @@ from Utilities.filecreator import create
BASKETS_BASE_DIRECTORY = "baskets"
CONFIG_FILE_DEFAULT_NAME = "config.json"
-Baskets_Base_Location = None
-Config_File_Location = None
if sys.platform in ("linux", "linux2", "darwin"):
filesystem_location = '/etc/iotcollector'
elif sys.platform == "win32":
filesystem_location = os.path.dirname(os.path.realpath(__file__))
+Baskets_Base_Location = None
+Config_File_Location = None
+Host_Identifier = None
+
def set_filesystem_location(new_location):
global filesystem_location
filesystem_location = new_location
-def init_file_system(new_configfile_location=None):
- global Baskets_Base_Location, Config_File_Location
+def init_file_system(host_identifier=None, new_configfile_location=None):
+ global Baskets_Base_Location, Config_File_Location, Host_Identifier
try:
Baskets_Base_Location = os.path.join(filesystem_location,
BASKETS_BASE_DIRECTORY)
@@ -32,6 +34,8 @@ def init_file_system(new_configfile_loca
else:
Config_File_Location = create_file_if_not_exists(
os.path.join(filesystem_location, CONFIG_FILE_DEFAULT_NAME),
hidden=False, init_text='[]')
+
+ Host_Identifier = host_identifier
except (Exception, OSError) as ex:
print >> sys.stderr, ex
sys.exit(1)
@@ -43,3 +47,7 @@ def get_baskets_base_location():
def get_configfile_location():
return Config_File_Location
+
+
+def get_host_identifier():
+ return Host_Identifier
diff --git a/clients/iotclient/src/Settings/mapiconnection.py
b/clients/iotclient/src/Settings/mapiconnection.py
--- a/clients/iotclient/src/Settings/mapiconnection.py
+++ b/clients/iotclient/src/Settings/mapiconnection.py
@@ -28,11 +28,15 @@ def mapi_create_stream(schema, stream, c
Connection.execute("CREATE SCHEMA " + schema + ";")
except:
pass
- sql_str = ''.join(["CREATE STREAM TABLE ", schema, ".", stream, " (",
columns, ");"])
- Connection.execute(sql_str)
+
+ try: # attempt to create te stream table
+ Connection.execute(''.join(["CREATE STREAM TABLE ", schema, ".",
stream, " (", columns, ");"]))
+ except:
+ pass
def mapi_flush_baskets(schema, stream, baskets):
- # this procedure does not work yet. Have to check it with Martin
- # Connection.execute(''.join(["CALL iot.push(\"", schema, "\",\"", stream,
"\",\"", baskets, "\");"]))
- pass
+ try:
+ Connection.execute(''.join(["CALL iot.push(\"", schema, "\",\"",
stream, "\",\"", baskets, "\");"]))
+ except:
+ pass
diff --git a/clients/iotclient/src/Streams/streams.py
b/clients/iotclient/src/Streams/streams.py
--- a/clients/iotclient/src/Streams/streams.py
+++ b/clients/iotclient/src/Streams/streams.py
@@ -1,16 +1,14 @@
import os
from collections import defaultdict
-from datatypes import TimestampType, DataValidationException
+from datatypes import TimestampType, TextType, DataValidationException
from flushing import TimeBasedFlushing, TupleBasedFlushing
from Settings.mapiconnection import mapi_create_stream, mapi_flush_baskets
-from Settings.filesystem import get_baskets_base_location
+from Settings.filesystem import get_baskets_base_location, get_host_identifier
from Settings.iotlogger import add_log
from Utilities.filecreator import create_file_if_not_exists,
get_hidden_file_name
from Utilities.readwritelock import RWLock
-IMPLICIT_TIMESTAMP_COLUMN_NAME = 'implicit_timestamp'
-
def represents_int(s):
try:
@@ -19,6 +17,23 @@ def represents_int(s):
except ValueError:
return False
+IMPLICIT_TIMESTAMP_COLUMN_NAME = 'implicit_timestamp'
+Timestamps_Handler = TimestampType(name=IMPLICIT_TIMESTAMP_COLUMN_NAME,
type="timestamp") # timestamp
+Create_SQL_array = [Timestamps_Handler.create_stream_sql()] # array for SQL
creation
+
+HOST_IDENTIFIER_COLUMN_NAME = 'host_identifier'
+Hostname_Bin_Value = None
+
+
+def init_streams_hosts():
+ global Hostname_Bin_Value
+
+ host_identifier = get_host_identifier()
+ if host_identifier is not None:
+ hosts_handler = TextType(name=HOST_IDENTIFIER_COLUMN_NAME,
type="text") # host_identifier
+ Hostname_Bin_Value = hosts_handler.process_values([host_identifier])
+ Create_SQL_array.append(hosts_handler.create_stream_sql())
+
class StreamException(Exception):
"""Exception fired when the validation of a stream insert fails"""
@@ -36,10 +51,10 @@ class DataCellStream(object):
self._tuples_in_per_basket = 0 # for efficiency
self._flush_method = flush_method # instance of StreamFlushingMethod
self._columns = columns # dictionary of name -> data_types
- self._timestamps_handler =
TimestampType(name=IMPLICIT_TIMESTAMP_COLUMN_NAME, type="timestamp") #
timestamp
self._validation_schema = validation_schema # json validation schema
for the inserts
self._monitor = RWLock() # baskets lock to protect files (the server
is multi-threaded)
self._base_path = os.path.join(get_baskets_base_location(),
schema_name, stream_name)
+
if not os.path.exists(self._base_path):
os.makedirs(self._base_path)
self._baskets_counter = 1
@@ -49,18 +64,21 @@ class DataCellStream(object):
for elem in dirs: # for each directory found, flush it
dir_path = os.path.join(self._base_path, str(elem))
mapi_flush_baskets(self._schema_name, self._stream_name,
dir_path)
- self._baskets_counter = max(dirs) + 1 # the current basket
number will be the next one
+ self._baskets_counter = max(dirs) + 1 # increment current
basket number
else:
self._baskets_counter = 1
self._current_base_path = os.path.join(self._base_path,
str(self._baskets_counter))
os.makedirs(self._current_base_path)
+
for key in self._columns.keys(): # create the files for the columns
and timestamp
create_file_if_not_exists(os.path.join(self._current_base_path,
key), hidden=True)
create_file_if_not_exists(os.path.join(self._current_base_path,
IMPLICIT_TIMESTAMP_COLUMN_NAME), hidden=True)
+ if Hostname_Bin_Value is not None:
+ create_file_if_not_exists(os.path.join(self._current_base_path,
HOST_IDENTIFIER_COLUMN_NAME), hidden=True)
if created: # when the stream is reloaded from the config file, the
create SQL statement is not sent
- column_string = ','.join([column.create_stream_sql() for column in
self._columns.values()])
- mapi_create_stream(self._schema_name, self._stream_name,
column_string)
+ sql_array = [column.create_stream_sql() for column in
self._columns.values()]
+ mapi_create_stream(self._schema_name, self._stream_name,
','.join(sql_array + Create_SQL_array))
def get_schema_name(self):
return self._schema_name
@@ -101,10 +119,14 @@ class DataCellStream(object):
self._baskets_counter += 1
self._current_base_path = os.path.join(self._base_path,
str(self._baskets_counter))
os.makedirs(self._current_base_path)
+
for key in self._columns.keys():
create_file_if_not_exists(os.path.join(self._current_base_path, key),
hidden=True)
create_file_if_not_exists(os.path.join(self._current_base_path,
IMPLICIT_TIMESTAMP_COLUMN_NAME),
hidden=True)
+ if Hostname_Bin_Value is not None:
+
create_file_if_not_exists(os.path.join(self._current_base_path,
HOST_IDENTIFIER_COLUMN_NAME),
+ hidden=True)
def time_based_flush(self, last=False):
self._monitor.acquire_write()
@@ -152,13 +174,17 @@ class DataCellStream(object):
# prepare variables outside the lock for more parallelism
total_tuples = len(new_data)
- bin_value = self._timestamps_handler.process_values([timestamp])
- timestamps_binary_array = ''.join([bin_value for _ in
xrange(total_tuples)])
+
+ timestamp_bin_value = Timestamps_Handler.process_values([timestamp])
+ timestamps_binary_array = ''.join([timestamp_bin_value for _ in
xrange(total_tuples)])
+
+ if Hostname_Bin_Value is not None:
+ hosts_binary_array = ''.join([Hostname_Bin_Value for _ in
xrange(total_tuples)])
+
+ # supposing that the flushing method never changes we can do this
outside the lock
is_flushing_tuple_based = isinstance(self._flush_method,
TupleBasedFlushing)
- # supposing that the flushing method never changes we can do this
outside the lock
self._monitor.acquire_write()
-
for key, inserts in transposed_data.iteritems(): # now write the
binary data
# open basket in binary mode and append the new entries
basket_fp =
open(get_hidden_file_name(os.path.join(self._current_base_path, key)), 'ab')
@@ -173,6 +199,13 @@ class DataCellStream(object):
time_basket_fp.flush()
time_basket_fp.close()
+ if Hostname_Bin_Value is not None: # write the host name if applicable
+ hosts_basket_fp =
open(get_hidden_file_name(os.path.join(self._current_base_path,
+
HOST_IDENTIFIER_COLUMN_NAME)), 'ab')
+ hosts_basket_fp.write(hosts_binary_array)
+ hosts_basket_fp.flush()
+ hosts_basket_fp.close()
+
self._tuples_in_per_basket += total_tuples
if is_flushing_tuple_based and self._tuples_in_per_basket >=
self._flush_method.limit:
self.flush_baskets(last=False)
diff --git a/clients/iotclient/src/main.py b/clients/iotclient/src/main.py
--- a/clients/iotclient/src/main.py
+++ b/clients/iotclient/src/main.py
@@ -2,8 +2,9 @@ import getopt
import sys
import threading
+from uuid import getnode as get_mac
from Settings import filesystem, iotlogger
-from Streams import streamscontext
+from Streams import streamscontext, streams
from Flask import restresources
from Flask.app import start_flask_iot_app, start_flask_admin_app
from Settings.mapiconnection import init_monetdb_connection
@@ -11,9 +12,11 @@ from Settings.mapiconnection import init
def main(argv):
try:
- opts, args = getopt.getopt(argv[1:], 'f:l:c:ih:ip:ah:ap:ch:cp:cd:cu',
- ['filesystem=', 'logfile=', 'configfile=',
'ihost=', 'iport=', 'ahost=', 'aport=',
- 'chostname=', 'cport=', 'cdatabase=',
'cuser='])
+ opts, args = getopt.getopt(argv[1:],
'f:l:c:u:n:ih:ip:ah:ap:dh:dp:dd:du',
+ ['filesystem=', 'logfile=', 'configfile=',
'useidentifier=', 'name='
+
'ihost=', 'iport=',
+ 'ahost=', 'aport=',
+ 'dhostname=', 'dport=', 'ddatabase=',
'duser='])
except getopt.GetoptError:
print >> sys.stderr, "Error while parsing the arguments!"
sys.exit(1)
@@ -28,7 +31,10 @@ def main(argv):
connection_port = 50000
connection_user = 'monetdb'
connection_database = 'iotdb'
+
new_configfile_location = None
+ use_host_identifier = False
+ host_identifier = None
for opt, arg in opts:
if opt in ('-f', '--filesystem'):
@@ -37,6 +43,10 @@ def main(argv):
iotlogger.set_logging_location(arg)
elif opt in ('-c', '--configfile'):
new_configfile_location = arg
+ elif opt in ('-u', '--useidentifier'):
+ use_host_identifier = bool(arg)
+ elif opt in ('-n', '--name'):
+ host_identifier = arg
elif opt in ('-ih', '--ihost'):
app_host = arg
@@ -47,18 +57,24 @@ def main(argv):
elif opt in ('-ap', '--aport'):
admin_port = int(arg)
- elif opt in ('-ch', '--chostname'):
+ elif opt in ('-dh', '--dhostname'):
connection_hostname = arg
- elif opt in ('-cp', '--cport'):
+ elif opt in ('-dp', '--dport'):
connection_port = int(arg)
- elif opt in ('-cu', '--cuser'):
+ elif opt in ('-du', '--duser'):
connection_user = arg
- elif opt in ('-cd', '--cdatabase'):
+ elif opt in ('-dd', '--ddatabase'):
connection_database = arg
+ if use_host_identifier and host_identifier is None: # get the machine MAC
address as default identifier
+ host_identifier = ':'.join(("%012X" % get_mac())[i:i + 2] for i in
range(0, 12, 2))
+ if not use_host_identifier: # in case of the user sets the
host_identifier but not the use_host_identifier flag
+ host_identifier = None
+
# WARNING The initiation order must be this!!!
iotlogger.init_logging() # init logging context
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
