tickHF:initial implementation of issue #44

(not yet tested)

components/tickHF/test/README.md

@@ -0,0 +1,28 @@
+### Executing tests
+
+Executing tests (assuming ec is deployed in the bin direcotory):
+
+- prepare env on linux:
+```bash
+KdbSystemDir> source bin/ec/components/tickHF/test/etc/env.sh
+```
+- prepare env on windows:
+```bash
+KdbSystemDir> bin\ec\components\tickHF\test\etc\env.bat
+```
+
+- execute tests:
+```bash
+> # start tests
+> yak start t.run
+> # check progress
+> yak log t.run
+> # check errors and test failures
+> yak err t.run
+```
+
+- inspect results once the tests are completed:
+```q  
+q)//on the t.run
+q).test.report[]
+```

components/tickHF/test/etc/access.cfg

@@ -0,0 +1,9 @@
+[technicalUser:tu]
+  pass = 0x5f5e005b4a5858
+  usergroups = admin
+#---------------------------- user groups ------------------------------------#
+[userGroup:admin]
+  [[ALL]]
+    namespaces = ALL
+    checkLevel = NONE
+#-----------------------------------------------------------------------------#

components/tickHF/test/etc/dataflow.cfg

@@ -0,0 +1,21 @@
+[template:marketData]
+  [[t0.tickHF]]
+  [[t1.tickHF]]
+  [[t0.rdb]]
+    subSrc = t0.tick      
+    eodClear = FALSE
+    eodPerform = FALSE
+  [[t1.rdb]]
+    subSrc = t1.tick      
+    eodClear = FALSE
+    eodPerform = FALSE
+
+#---------------------------- market data tables -----------------------------#
+[table:quote]
+  template = marketData
+  model = time(TIME), sym(SYMBOL), bid(FLOAT), bidSize(LONG), ask(FLOAT), askSize(LONG), flag(STRING)
+
+#---------------------------- system tables ----------------------------------#
+[sysTable:dummy]
+  modelSrc = NULL
+#-----------------------------------------------------------------------------#

components/tickHF/test/etc/env.bat

@@ -0,0 +1,34 @@
+@ECHO OFF
+IF NOT EXIST BIN GOTO NOBIN
+
+SET EC_SYS_PATH=%CD%
+
+REM  ---- q enviroment ----
+IF EXIST %EC_SYS_PATH%\bin\q\w32\q.exe SET QHOME=%EC_SYS_PATH%\bin\q
+IF EXIST %EC_SYS_PATH%\bin\q\w64\q.exe SET QHOME=%EC_SYS_PATH%\bin\q
+SET PATH=%QHOME%\w64;%QHOME%\w32;%PATH%
+WHERE /q q.exe || GOTO NOQ
+REM in case of q64bit - it is recommended to SET QLIC as following: QLIC=%EC_SYS_PATH%\etc
+
+REM ---- ec enviroment ----
+SET EC_QSL_PATH=%EC_SYS_PATH%/bin/ec/libraries/qsl/
+SET EC_ETC_PATH=%EC_SYS_PATH%/bin/ec/components/tickHF/test/etc/
+
+SET EC_SYS_ID="tickHFTest"
+SET EC_SYS_TYPE="FUNC_TEST"
+
+REM ---- yak enviroment ----
+SET YAK_PATH=%EC_SYS_PATH%\bin\yak
+SET YAK_OPTS=-c %EC_ETC_PATH%/system.cfg -s %EC_SYS_PATH%/data/test/%EC_SYS_ID%/yak/yak.status -l %EC_SYS_PATH%/log/test/%EC_SYS_ID%/yak/yak.log
+SET PATH=%YAK_PATH%;%PATH%
+
+ECHO env for system %EC_SYS_ID%/%EC_SYS_TYPE% loaded
+
+
+GOTO END
+:NOBIN
+ECHO Expected to find %EC_SYS_PATH%\bin directory. Expecting to run env.bat from the top level ec system directory.
+GOTO END
+:NOQ
+ECHO Missing q binary. Please install q in %EC_SYS_PATH%\bin\q\ directory, see ec\tutorial\Installation.md for instructions.
+:END

components/tickHF/test/etc/env.sh

@@ -0,0 +1,41 @@
+if [ ! -d "bin" ]; then
+  echo "Expecting bin subdirectory" && return
+fi
+export EC_SYS_PATH=${PWD}
+
+# ---- cmd prompt decoration ---- #
+EC_SYS_ID="tickHFTest"
+EC_SYS_TYPE="FUNC_TEST"
+export EC_SYS_ID EC_SYS_TYPE
+PS1='[${EC_SYS_ID}(${EC_SYS_TYPE})][\u@\h:\w]\$ '
+export PS1
+
+# ---- ec environment ---- #
+EC_QSL_PATH=${EC_SYS_PATH}/bin/ec/libraries/qsl/
+EC_ETC_PATH=${EC_SYS_PATH}/bin/ec/components/tickHF/test/etc/
+export EC_QSL_PATH EC_ETC_PATH
+
+# ---- yak environment ---- #
+YAK_PATH=$EC_SYS_PATH/bin/yak/
+YAK_OPTS="-c $EC_ETC_PATH/system.cfg -v less -s $EC_SYS_PATH/data/test/${EC_SYS_ID}/yak/yak.status -l $EC_SYS_PATH/log/test/${EC_SYS_ID}/yak/yak.log"
+export YAK_OPTS
+. $YAK_PATH/yak_complete_bash.sh
+NEWPATH=$YAK_PATH
+
+# ---- q environment ---- #
+# search for q binary in bin/q/ (cover MacOS and Linux)
+if [ "$(uname)" == "Darwin" ]; then OS_LETTER=m; else OS_LETTER=l; fi
+if [ -f "$EC_SYS_PATH/bin/q/${OS_LETTER}32/q" ]; then QHOME=$EC_SYS_PATH/bin/q/ && NEWPATH=$QHOME/${OS_LETTER}32/:$NEWPATH; fi
+if [ -f "$EC_SYS_PATH/bin/q/${OS_LETTER}64/q" ]; then QHOME=$EC_SYS_PATH/bin/q/ && NEWPATH=$QHOME/${OS_LETTER}64/:$NEWPATH; fi
+# QLIC set to etc/, in case of q64, k4.lic should be placed in $QLIC directory
+QLIC=$EC_SYS_PATH/etc/
+export QHOME QLIC
+
+# ---- path ---- #
+export OLDPATH=${OLDPATH:-$PATH}
+PATH=$NEWPATH:$OLDPATH
+export PATH
+#check if q is available on PATH
+which q >/dev/null || echo "Missing q binary. Please install q in $EC_SYS_PATH/bin/q/ directory, see ec/tutorial/Installation.md for instructions."
+
+echo env for system ${EC_SYS_ID}/${EC_SYS_TYPE} loaded

components/tickHF/test/etc/system.cfg

@@ -0,0 +1,55 @@
+export = libPath, dllPath, logDest, logPath, logRotate, logLevel, eventPath
+#---------------------------------- paths ------------------------------------#
+etcPath = ${EC_SYS_PATH}/bin/ec/components/hdbWriter/test/etc/
+binPath = ${EC_SYS_PATH}/bin/ec/components/${EC_COMPONENT_PKG}
+libPath = ${EC_SYS_PATH}/bin/ec/components/${EC_COMPONENT_PKG}, ${EC_SYS_PATH}/bin/ec/libraries/
+dllPath = ${EC_SYS_PATH}/bin/ec/components/${EC_COMPONENT_PKG}, ${EC_SYS_PATH}/bin/ec/libraries/
+dataPath =  ${EC_SYS_PATH}/data/test/tickHFTest/${EC_COMPONENT_ID}
+logPath =   ${EC_SYS_PATH}/log/test/tickHFTest/${EC_COMPONENT_ID}
+eventPath = ${EC_SYS_PATH}/data/test/tickHFTest/shared/events/
+#---------------------------------- process ----------------------------------#
+basePort = 18000
+startWait = 1
+stopWait = 1
+cpuAffinity = 0
+#---------------------------------- logging ----------------------------------#
+logLevel = INFO
+logDest = FILE,STDERR
+logRotate = 01:00:00
+#---------------------------------- housekeeping -----------------------------#
+housekeeping = ()
+#---------------------------------- components -------------------------------#
+[group:t]
+  [[t.run]]
+    command = "q qtestRunner.q"
+    type = q:qtest/qtestRunner
+    port = ${basePort}
+    memCap = 10000
+    testFiles = ${EC_SYS_PATH}/bin/ec/components/tickHF/test/tickHF_functionalTests.q
+    testNamespaces = .testTickHF
+
+  [[t0.tickHF]]
+    command = "q tickHF.q"
+    type = q:tickHF/tickHF
+    port = ${basePort} + 1
+    memCap = 10000
+
+  [[t1.tickHF]]
+    command = "q tickHF.q"
+    type = q:tickHF/tickHF
+    port = ${basePort} + 2
+    memCap = 10000  
+
+
+  [[t0.rdb]]
+    command = "q rdb.q"
+    type = q:rdb/rdb
+    port = ${basePort} + 11
+    memCap = 10000
+
+  [[t1.rdb]]
+    command = "q rdb.q"
+    type = q:rdb/rdb
+    port = ${basePort} + 12
+    memCap = 10000
+#-----------------------------------------------------------------------------#

components/tickHF/test/tickHF_functionalTests.q

@@ -0,0 +1,79 @@
+/L/ Copyright (c) 2011-2014 Exxeleron GmbH
+/L/
+/L/ Licensed under the Apache License, Version 2.0 (the "License");
+/L/ you may not use this file except in compliance with the License.
+/L/ You may obtain a copy of the License at
+/L/
+/L/   http://www.apache.org/licenses/LICENSE-2.0
+/L/
+/L/ Unless required by applicable law or agreed to in writing, software
+/L/ distributed under the License is distributed on an "AS IS" BASIS,
+/L/ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+/L/ See the License for the specific language governing permissions and
+/L/ limitations under the License.
+
+/A/ DEVnet: Joanna Wdowiak
+/V/ 3.0
+
+// Functional tests of the tickHF component
+// See README.md for details
+
+.testTickHF.testSuite:"tickHF functional tests";
+
+.testTickHF.setUp:{
+  };
+
+.testTickHF.tearDown:{
+  .test.stop `t0.tickHF`t1.tickHF`t0.rdb`t1.rdb;
+  .test.clearProcDir `t0.tickHF`t1.tickHF`t0.rdb`t1.rdb;
+  };
+
+//----------------------------------------------------------------------------//
+.testTickHF.genQuote:{[cnt]
+  ([]time:`time$til cnt; sym:cnt#`aaa`bbb;bid:`float$til cnt; bidSize:til cnt;ask:`float$til cnt; askSize:til cnt; flag:cnt?("flagA";"flagB"))
+  };
+
+//----------------------------------------------------------------------------//
+//                               test tickHF                                  //
+//----------------------------------------------------------------------------//
+.testTickHF.test.tickHF_start_with_ok_journal:{[]
+  // insert data to journal
+  data:.testTickHF.genQuote 5;
+  jrn:.Q.dd[.cr.getCfgField[`t0.tickHF;`group;`dataPath];`$"t0.tickHF",string[.z.d]];
+  .[jrn;();:;()];
+  h:hopen jrn;
+  h enlist (`jUpd;`quote;value flip data);
+  // run tick
+  .test.start[`t0.tickHF];
+  // sub from rdb
+  .test.start[`t0.rdb];
+  .assert.match["rdb contains loaded quote table";rdb"select from quote"; `date`time xasc tradeChunk,tradeChunk];
+  };
+
+.testTickHF.test.tickHF_start_with_bad_journal_abort:{[]
+
+  };
+
+.testTickHF.test.tickHF_start_with_bad_journal_archive:{[]
+
+  };
+
+.testTickHF.test.tickHF_start_with_bad_journal_truncate:{[]
+
+  };
+
+.testTickHF.test.tickHF_late_ticks_start_with_ok_journal:{[]
+
+  };
+
+.testTickHF.test.tickHF_late_ticks_start_with_bad_journal_abort:{[]
+
+  };
+
+.testTickHF.test.tickHF_late_ticks_start_with_bad_journal_archive:{[]
+
+  };
+
+.testTickHF.test.tickHF_late_ticks_start_with_bad_journal_truncate:{[]
+
+  };

components/tickHF/tickHF.q

@@ -78,6 +78,7 @@ system"l ",getenv[`EC_QSL_PATH],"/sl.q";
 /------------------------------------------------------------------------------/
 .sl.lib[`$"qsl/u"];
 .sl.lib[`$"qsl/timer"];
+.sl.lib[`$"qsl/os"];
 .sl.lib["cfgRdr/cfgRdr"];
 
 /------------------------------------------------------------------------------/
@@ -92,11 +93,13 @@ system"l ",getenv[`EC_QSL_PATH],"/sl.q";
 /F/ Function opens/creates the journal file for specified date and initializes <.u.i> with proper value
 /P/ x - date
 /R/ journal handle
+
 ld:{
   if[not type key L::`$(-10_string L),string x;
     .[L;();:;()]
     ];
   j::i::-11!(-2;L);
+  if[0<=type i;.tickHF.p[.tickHF.cfg.handleBadJrn;.u.L;.u.i;`today]];
   h:hopen L;
   .log.info[`tickHF] "Loading journal file:`",string[L],", handle:",string[h], ", entries:", string[i];
   :h;
@@ -110,6 +113,7 @@ nextday.ld:{
     .[nextday.L;();:;()]
     ];
   nextday.j::nextday.i::-11!(-2;nextday.L);
+  if[0<=type nextday.i;.tickHF.p[.tickHF.cfg.handleBadJrn;.u.nextday.L;.u.nextday.i;`nextday]];
   h:hopen nextday.L;
   .log.info[`tickHF] "Loading journal file:`",string[nextday.L],", handle:",string[h], ", entries:", string[nextday.i];
   :h;
@@ -313,12 +317,41 @@ p.zeroLatencyMode:{[]
   @[;`sym;`g#]each tabs;
   };
 
+// jrnPath:.u.L
+// i:(1;2)
+.tickHF.p.ABORT:{[jrnPath;i;day]
+  .log.error[`tickHF]`errMsg`validChunks`validPartSize`totalJrnSize!("Corrupted journal file ",string[jrnPath],". Process will be stopped";i[0];i[1];hcount jrnPath);
+  .log.info[`tickHF]"Please archive or truncate journal ",string[jrnPath]," to lenght " ,string[i 1] ," bytes manually and restart the process";
+  exit 1;
+  };
+
+.tickHF.p.ARCHIVE_BAD_JRN:{[jrnPath;i;day]
+  archivePath:(1_string jrnPath),"_corrupted_after_",(string i 1);
+  .log.error[`tickHF]`errMsg`validChunks`validPartSize`totalJrnSize!("Corrupted journal file ",string[jrnPath];i[0];i[1];hcount jrnPath);
+  .log.info[`tickHF]"Journal is archived to ",archivePath, " and new one will be created";
+  .log.info[`tickHF]"Please note that messages from the invalid journal will be dropped. Afterwards it can be manually injected into the running system";
+  .os.move[1_string jrnPath;archivePath];
+  .[jrnPath;();:;()];
+  $[`today~day;.u.j::.u.i::-11!(-2;jrnPath);.u.nextday.j::.u.nextday.i::-11!(-2;jrnPath)];
+  };
+
+.tickHF.p.TRUNCATE_BAD_TAIL:{[jrnPath;i;day]
+  archivePath:(1_string jrnPath),"_corrupted_after_",(string i 1);
+  .log.error[`tickHF]`errMsg`validChunks`validPartSize`totalJrnSize!("Corrupted journal file ",string[jrnPath];i[0];i[1];hcount jrnPath);
+  .log.info[`tickHF]"Journal is archived to ",archivePath;
+  .os.cp[1_string jrnPath;archivePath];
+  .log.info[`tickHF]"Journal is truncated to size ",string[i 1]," and new messages will be appened to the existing journal";
+  .os.truncate[1_string jrnPath;i 1];
+  $[`today~day;.u.j::.u.i::-11!(-2;jrnPath);.u.nextday.j::.u.nextday.i::-11!(-2;jrnPath)];
+  };
+
 /==============================================================================/
 .sl.main:{[flags]
   (set) ./:                        model:.cr.getModel[`THIS];
   .tickHF.cfg.dataPath:            1_string .cr.getCfgField[`THIS;`group;`cfg.dataPath];
   .tickHF.cfg.jrnPrefix:           .cr.getCfgField[`THIS;`group;`cfg.jrnPrefix];
   .tickHF.cfg.aggrInterval:       .cr.getCfgField[`THIS;`group;`cfg.aggrInterval];
+  .tickHF.cfg.handleBadJrn:       .cr.getCfgField[`THIS;`group;`cfg.handleBadJrn];
   // if null - zero mode
   $[null .tickHF.cfg.aggrInterval;.u.p.zeroLatencyMode[];.u.p.aggrMode[]];
   .sl.libCmd[];

components/tickHF/tickHF.qsd

@@ -8,6 +8,8 @@
   #/C/ aggregation interval, messages are cached and published on given time interval
   #/E/ cfg.aggrInterval = 100
   cfg.aggrInterval = <type(INT), default(NULL)>
+  #/C/ action to handle corrupted journal
+  cfg.handleBadJrn = <type(SYMBOL), in(ABORT, ARCHIVE_BAD_JRN, TRUNCATE_BAD_TAIL), default(ABORT)>
 [table]
 [sysTable]  
 [user]

libraries/qsl/os.q

@@ -172,6 +172,21 @@
 /P/ p:STRING - a path
 .os.p.W.slash:{[p] ssr[p;"/";"\\"]};
 
+/F/ cuts a given file at given position, Linux version
+/P/ file:STRING - path to file
+/P/ size:LONG - bytes to take
+.os.p.L.truncate:{[file;size]
+  file:.os.remSlash .os.p.fixPath file;
+  system "truncate -s ",(string size)," ",file;
+  };
+
+/F/ cuts a given file at given position, Windows version - to be implemented
+/P/ file:STRING - path to file
+/P/ size:LONG - bytes to take
+.os.p.W.truncate:{[file;size] 
+  '`$"the truncate function is not implemented on Windows OS";
+  };
+
 /F/ a wrapper for the system calls. The standard error is redirected to a disk file and thrown as signal.
 /P/ cmd:STRING - command string
 .os.system:{[origin;cmd]
@@ -232,5 +247,10 @@
 /P/ t:LONG - time in millisecods. On Windows, this parameter is rounded UP to the nearest multiple of 1000.
 .os.sleep:{[t] };
 
+/F/ truncate file to given sie. There is no implementation for Windows
+/P/ file:STRING - path to file
+/P/ size:LONG - bytes to take
+.os.truncate:{[file;size] };
+
 /--- initialization
 $["w"~first string .z.o;.os,:.os.p.W;.os,:.os.p.L];