-/* ------------------------------------------------\r
- Copyright 2014 AT&T Intellectual Property\r
- Licensed under the Apache License, Version 2.0 (the "License");\r
- you may not use this file except in compliance with the License.\r
- You may obtain a copy of the License at\r
- \r
- http://www.apache.org/licenses/LICENSE-2.0\r
- \r
- Unless required by applicable law or agreed to in writing, software\r
- distributed under the License is distributed on an "AS IS" BASIS,\r
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
- See the License for the specific language governing permissions and\r
- limitations under the License.\r
- ------------------------------------------- */\r
-#include <app.h>\r
-#include <stdlib.h>\r
-#include <string.h>\r
-#include <stdio.h>\r
-#include <unistd.h>\r
-#include <signal.h>\r
-#include <time.h>\r
-#include <string.h>\r
-#include <sys/time.h>\r
-#include <sys/stat.h>\r
-#include <sys/types.h>\r
-#include <sys/socket.h>\r
-#include <netinet/in.h>\r
-\r
-#include "gsconfig.h"\r
-#include "gstypes.h"\r
-#include <schemaparser.h>\r
-#include <gshub.h>\r
-\r
-#define MAXSTRLEN 256\r
-#define MAXNUMFIELDS 64\r
-\r
-//#define GSBZIP2\r
-\r
-#ifdef GSBZIP2\r
-#include "bzlib.h"\r
-#endif\r
-\r
-#ifdef ZLIB\r
-#include "zlib.h"\r
-#endif\r
-\r
-gs_sp_t me; /* saved copy argv[0] */\r
-\r
-struct FTA_state {\r
- FTAID fta_id;\r
- gs_schemahandle_t schema;\r
- gs_sp_t asciischema;\r
- gs_int32_t numfields;\r
- gs_int32_t timefieldoffset;\r
-};\r
-\r
-struct gsasciiprint_state {\r
- // configuration state\r
- gs_uint32_t bufsz;\r
- gs_int32_t stream;\r
- gs_int32_t flush;\r
- gs_int32_t parserversion;\r
- gs_int32_t compressed;\r
- gs_int32_t verbose;\r
- gs_int32_t notemp;\r
- gs_sp_t timefield;\r
- gs_int32_t interval;\r
- gs_int32_t quitcnt;\r
- gs_int32_t quittime;\r
- gs_int32_t numfields;\r
- gs_sp_t extension;\r
- gs_sp_t query;\r
- gs_int32_t remote_print;\r
- \r
- \r
- // runtime state\r
- struct FTA_state fs;\r
-};\r
-\r
-\r
-void hand(int iv) {\r
- ftaapp_exit();\r
- gslog(LOG_NOTICE, "exiting via signal handler %d...\n", iv);\r
- exit(0);\r
-}\r
-\r
-void timeouthand(int iv) {\r
- ftaapp_exit();\r
- // if (s->verbose!=0) fprintf(stderr, "exiting because of timeout...\n");\r
- exit(0);\r
-}\r
-\r
-\r
-static void print_usage_exit(gs_sp_t reason) {\r
- fprintf(stderr,\r
- "%s::error: %s\n"\r
- "%s::usage: %s -r <int> -v -f -s -z -c <int> -q <int> -b <field> -t <int> -e <string> <gshub-hostname>:<gshub-port> <gsinstance_name> <query_name> <parameters>\n"\r
- "\t-v makes output verbose\n"\r
- "\t-f flushes each tuple individually\n"\r
- "\t-h <int> print within the hfta int identifies how many output streams based on the SubInterface field\n"\r
- "\t-r sets the ringbuffer size default is 8MB\n"\r
- "\t-s uses %s in streaming mode cannot be used with -v -z -b -t -e\n"\r
- "\t-c <int> indicates that %s should terminate after <int> tuples.\n"\r
- "\t-q <int> indicates that %s should terminate after <int> seconds.\n"\r
- "\t-z the output is compressed with gzip \n"\r
- "\t-b <field> identifies the field which is increasing and is used to\n"\r
- "\t\tbin output files. Field has to be of type uint.\n"\r
- "\t\tThe default is the UNIX system time\n"\r
- "\t-t <int> specifies the number the <field> specified with -b\n"\r
- "\t\thas to increase before a new output file is used\n"\r
- "\t\tthe default is 60 for 1 minute if the default UNIX time is used\n"\r
- "\t-e <string> identifies the file string extension used for the \n"\r
- "\t\toutput file. The output file always starts with the current value\n"\r
- "\t\tof the b field (or UNIX time) the default extension is .txt \n"\r
- "\t\twithout the -z option and .txt.gz with the -z option\n"\r
- "\t<query_name> specifies the query which will be\n"\r
- "\t\t instanciated.\n"\r
- "\t<parameters> sets the parameters for the query\n"\r
- , me, reason, me, me, me,me,me);\r
- exit(1);\r
-}\r
-\r
-\r
-\r
-\r
-static void init(gs_int32_t argc, gs_sp_t argv[], struct gsasciiprint_state *s) {\r
- void *pblk;\r
- gs_int32_t x, y, schema, pblklen, lcv;\r
- gs_int32_t n_actual_param;\r
- gs_int32_t n_expected_param;\r
- gs_sp_t c;\r
- gs_int8_t name[1024];\r
- \r
- sprintf(name,"gsgdatprint:%s",argv[0]);\r
- \r
- gsopenlog(name);\r
- \r
- \r
- if (s->verbose!=0) gslog(LOG_DEBUG,"Initializing gscp\n");\r
- if (ftaapp_init(s->bufsz)!=0) {\r
- gslog(LOG_EMERG,"%s::error:could not initialize gscp\n", me);\r
- exit(1);\r
- }\r
- \r
- signal(SIGTERM, hand);\r
- signal(SIGINT, hand);\r
- signal(SIGPIPE, hand);\r
- \r
- if (s->verbose!=0) gslog(LOG_DEBUG,"Initializing FTAs\n");\r
- \r
- schema = ftaapp_get_fta_schema_by_name(argv[0]);\r
- if (schema < 0) {\r
- gslog(LOG_EMERG,"%s::error:can't get fta '%s' schema\n", me ,argv[0]);\r
- exit(1);\r
- }\r
- n_expected_param = ftaschema_parameter_len(schema);\r
- if (n_expected_param == 0) {\r
- pblk = 0;\r
- pblklen = 0;\r
- if (s->verbose) gslog(LOG_DEBUG,"[query does not have any params]\n");\r
- } else {\r
- n_actual_param = argc-1;\r
- if(n_actual_param < n_expected_param){\r
- gslog(LOG_EMERG,"Error, %d query parameters expected, %d provided.\n",n_expected_param, n_actual_param);\r
- exit(1);\r
- }\r
- /* parse the params */\r
- for (lcv = 1 ; lcv < argc ; lcv++) {\r
- char *k, *e;\r
- int rv;\r
- k = argv[lcv];\r
- e = k;\r
- while (*e && *e != '=') e++;\r
- if (*e == 0) {\r
- gslog(LOG_EMERG,"param parse error '%s' (fmt 'key=val')\n",\r
- argv[lcv]);\r
- exit(1);\r
- }\r
- *e = 0;\r
- rv = ftaschema_setparam_by_name(schema, k, e+1, strlen(e+1));\r
- *e = '=';\r
- if (rv < 0) {\r
- gslog(LOG_EMERG,"param setparam error '%s' (fmt 'key=val')\n",\r
- argv[lcv]);\r
- exit(1);\r
- }\r
- }\r
- if (ftaschema_create_param_block(schema, &pblk, &pblklen) < 0) {\r
- gslog(LOG_EMERG, "ftaschema_create_param_block failed!\n");\r
- exit(1);\r
- }\r
- }\r
- ftaschema_free(schema); /* XXXCDC */\r
- \r
- \r
- if (s->remote_print>=0) {\r
- s->fs.fta_id=ftaapp_add_fta_print(s->query,pblk?0:1,pblk?0:1,0,pblklen,pblk,"./data",s->extension,s->timefield,"SubInterface",s->interval,s->remote_print);\r
- if (s->fs.fta_id.streamid==0){\r
- gslog(LOG_EMERG,"%s::error:could not initialize fta_print %s\n",\r
- me,s->query);\r
- exit(1);\r
- }\r
- again:\r
- // wait forever\r
- sleep(60);\r
- goto again;\r
- }\r
- \r
- \r
- s->fs.fta_id=ftaapp_add_fta(s->query,pblk?0:1,pblk?0:1,0,pblklen,pblk);\r
- if (s->fs.fta_id.streamid==0){\r
- gslog(LOG_EMERG,"%s::error:could not initialize fta %s\n",\r
- me,s->query);\r
- exit(1);\r
- }\r
- /* XXXCDC: pblk is malloc'd, should we free it? */\r
- \r
- if ((c=ftaapp_get_fta_ascii_schema_by_name(s->query))==0){\r
- gslog(LOG_EMERG,"%s::error:could not get ascii schema for %s\n",\r
- me,s->query);\r
- exit(1);\r
- }\r
- \r
- //ftaapp_get_fta_ascii_schema_by_name uses static buffer so make a copy\r
- s->fs.asciischema=strdup(c);\r
- \r
- \r
- // Set parser version here\r
- s->parserversion=get_schemaparser_version();\r
- \r
- if ((s->fs.schema=ftaapp_get_fta_schema(s->fs.fta_id))<0) {\r
- gslog(LOG_EMERG,"%s::error:could not get schema for query\n",\r
- me,s->query);\r
- exit(1);\r
- }\r
- \r
- // Use all available fields\r
- if ((s->fs.numfields=ftaschema_tuple_len(s->fs.schema))<0) {\r
- gslog(LOG_EMERG,"%s::error:could not get number of fields for query %s\n",\r
- me,s->query);\r
- exit(1);\r
- }\r
- if (s->timefield!=0) {\r
- if ((s->fs.timefieldoffset=ftaschema_get_field_offset_by_name(\r
- s->fs.schema,s->timefield))<0) {\r
- gslog(LOG_EMERG,"%s::error:could not get "\r
- "offset for timefield %s in query %s\n",\r
- me,s->timefield,s->query);\r
- exit(1);\r
- }\r
- \r
- if (ftaschema_get_field_type_by_name(\r
- s->fs.schema,s->timefield)!=UINT_TYPE) {\r
- gslog(LOG_EMERG,"%s::error: illegal type for timefield "\r
- "%s in query %s UINT expected\n",\r
- me,s->timefield,s->query);\r
- exit(1);\r
- }\r
- }\r
-}\r
-\r
-\r
-static void process_data(struct gsasciiprint_state *s)\r
-{\r
- gs_int32_t x;\r
- gs_int32_t y;\r
- gs_int32_t z;\r
- gs_int32_t tb;\r
- gs_uint32_t nsz;\r
- FTAID rfta_id;\r
- gs_uint32_t rsize;\r
- gs_int8_t rbuf[2*MAXTUPLESZ];\r
- gs_int32_t problem;\r
- \r
- gs_int32_t ctb=0;\r
- gs_int8_t fname[1024];\r
- gs_int8_t tmpname[1024];\r
- gs_int8_t command[1024];\r
- gs_int8_t topb[1024];\r
- gs_int32_t rcnt=0;\r
- gs_retval_t code;\r
-#ifdef GSBZIP2\r
- int bzerror;\r
- BZFILE * b;\r
- int abandon=0;\r
- unsigned int bytes_in;\r
- unsigned int bytes_out;\r
-#endif\r
-#ifdef ZLIB\r
- gzFile of=0;\r
-#else\r
- FILE * of=0;\r
-#endif\r
- \r
- \r
- tmpname[0]=0;\r
- \r
- if (s->verbose!=0) gslog(LOG_INFO,"Getting Data for %s",s->query);\r
- \r
- sprintf(&topb[0],"GDAT\nVERSION:%u\nSCHEMALENGTH:%lu\n",\r
- s->parserversion,strlen(s->fs.asciischema)+1);\r
- if (s->stream!=0) {\r
- of=stdout;\r
- // need to get ASCII version of schema\r
- fwrite(&topb[0],strlen(topb),1,of);\r
- fwrite(s->fs.asciischema,strlen(s->fs.asciischema)+1,1,of);\r
- }\r
- \r
- if (s->quittime !=0 ) {\r
- signal(SIGALRM,timeouthand);\r
- alarm(s->quittime);\r
- }\r
- \r
- \r
- while((code=ftaapp_get_tuple(&rfta_id,&rsize,rbuf,2*MAXTUPLESZ,0))>=0) {\r
- rcnt++;\r
- if ((s->notemp==1) && (code==2)) continue;\r
- if (((s->quitcnt>0) && (rcnt>s->quitcnt))||ftaschema_is_eof_tuple(s->fs.schema, rbuf)) {\r
- if (s->verbose!=0)\r
- gslog(LOG_EMERG, "exiting reached tuple limit or all data has been proccessed\n");\r
- if (of!=0) {\r
- fclose(of);\r
- \r
- if (s->compressed) {\r
- system(command);\r
- }\r
- rename(tmpname,fname);\r
- }\r
- exit(0);\r
- }\r
- if (((code==0) || (code==2))&&(s->fs.fta_id.streamid==rfta_id.streamid)) {\r
- if (s->stream==0) {\r
- if (s->timefield!=0) {\r
- tb=fta_unpack_uint(rbuf,\r
- rsize,\r
- s->fs.timefieldoffset,\r
- &problem);\r
- } else {\r
- tb=time(0);\r
- }\r
- if ((ctb+s->interval)<=tb) {\r
- gsstats();\r
- if (of!=0) {\r
-#ifdef GSBZIP2\r
- BZ2_bzWriteClose(&bzerror, b, abandon, &bytes_in, &bytes_out );\r
- if (bzerror!=BZ_OK) {\r
- gslog(LOG_EMERG,"Could not bz close file .. EXITING\n");\r
- exit(0);\r
- }\r
-#endif\r
-#ifdef ZLIB\r
- gzclose(of);\r
-#else\r
- fclose(of);\r
-#endif\r
-#ifndef GSBZIP2\r
- if (s->compressed) {\r
- system(command);\r
- }\r
-#endif\r
- rename(tmpname,fname);\r
- }\r
- while((ctb+s->interval)<=tb) {\r
- if (ctb==0) {\r
- ctb=(tb/s->interval)*s->interval;\r
- } else {\r
- ctb=ctb+s->interval;\r
- }\r
- }\r
-#ifdef ZLIB\r
- sprintf(tmpname,"%u%s.gz.tmp",ctb,s->extension);\r
- sprintf(fname,"%u%s.gz",ctb,s->extension);\r
-#else\r
- sprintf(tmpname,"%u%s.tmp",ctb,s->extension);\r
- sprintf(fname,"%u%s",ctb,s->extension);\r
-#endif\r
-#ifndef GSBZIP2\r
- if (s->compressed) {\r
- sprintf(command,"gzip -S .tmpgz %s ; mv %s.tmpgz %s",tmpname,tmpname,tmpname);\r
- }\r
-#endif\r
-#ifdef ZLIB\r
- if ((of=gzopen(tmpname,"wb"))==0) {\r
- gslog(LOG_EMERG,"Could not open file \"%s\".. EXITING\n",\r
- tmpname);\r
- exit(0);\r
- }\r
-#else\r
- if ((of=fopen(tmpname,"w"))==0) {\r
- gslog(LOG_EMERG,"Could not open file \"%s\".. EXITING\n",\r
- tmpname);\r
- exit(0);\r
- }\r
-#endif\r
-#ifdef GSBZIP2\r
- if (s->compressed) {\r
- b=BZ2_bzWriteOpen(&bzerror,of,5,0,30);\r
- if (bzerror!=BZ_OK) {\r
- gslog(LOG_EMERG,"Could not bz open file \"%s\".. EXITING\n",\r
- tmpname);\r
- exit(0);\r
- }\r
- }\r
-#endif\r
- \r
- if (s->compressed) {\r
-#ifdef GSBZIP2\r
- BZ2_bzWrite ( &bzerror, b, &topb[0],strlen(topb));\r
- BZ2_bzWrite ( &bzerror, b, s->fs.asciischema,strlen(s->fs.asciischema)+1);\r
-#endif\r
- } else {\r
- // need to get ASCII version of schema\r
-#ifdef ZLIB\r
- gzwrite(of,&topb[0],strlen(topb));\r
- gzwrite(of,s->fs.asciischema,strlen(s->fs.asciischema)+1);\r
-#else\r
- fwrite(&topb[0],strlen(topb),1,of);\r
- fwrite(s->fs.asciischema,strlen(s->fs.asciischema)+1,1,of);\r
-#endif\r
- }\r
- }\r
- }\r
- if (code==0) {\r
- nsz=htonl(rsize);\r
- if (s->compressed) {\r
-#ifdef GSBZIP2\r
- BZ2_bzWrite ( &bzerror, b,&nsz,sizeof(gs_uint32_t) );\r
- BZ2_bzWrite ( &bzerror, b,rbuf,rsize);\r
-#endif\r
- } else {\r
-#ifdef ZLIB\r
- gzwrite(of,&nsz,sizeof(gs_uint32_t));\r
- gzwrite(of,rbuf,rsize);\r
- }\r
-#else\r
- if (fwrite(&nsz,sizeof(gs_uint32_t),1,of)!=1) {\r
- ftaapp_exit();\r
- if (s->verbose!=0)\r
- gslog(LOG_EMERG,"Could not write to output\"%s\".. EXITING\n",\r
- tmpname);\r
- exit(0);\r
- }\r
- \r
- if (fwrite(rbuf,rsize,1,of)!=1) {\r
- ftaapp_exit();\r
- if (s->verbose!=0)\r
- gslog(LOG_EMERG,"Could not write to output\"%s\".. EXITING\n",\r
- tmpname);\r
- exit(0);\r
- }\r
- if ((s->stream!=0) || (s->flush!=0)) {\r
- fflush(of);\r
- }\r
- }\r
-#endif\r
- }\r
- }\r
-}\r
-}\r
-\r
-\r
-\r
-int main(int argc, char** argv) {\r
- \r
- struct gsasciiprint_state s;\r
- gs_retval_t ch;\r
- endpoint gshub;\r
- endpoint dummyep;\r
- gs_uint32_t tip1,tip2,tip3,tip4;\r
- gs_sp_t instance_name;\r
- \r
- me = argv[0];\r
- \r
- /* initialize host library and the sgroup */\r
- \r
- if (argc<=1) {\r
- print_usage_exit("Not enough arguments");\r
- }\r
- \r
- /* parse args */\r
- bzero(&s, sizeof(s));\r
- s.interval = 60; /* default */\r
- s.quittime=0;\r
- s.quitcnt=-1;\r
- s.remote_print=-1;\r
- s.bufsz=8*1024*1024;\r
- \r
- while ((ch = getopt(argc, argv, "nr:b:e:t:c:q:sfvzmh:")) != -1) {\r
- switch(ch) {\r
- case 'r':\r
- s.bufsz=atoi(optarg);\r
- break;\r
- case 'c':\r
- s.quitcnt = atoi(optarg);\r
- break;\r
- case 'q':\r
- s.quittime = atoi(optarg);\r
- break;\r
- case 'b':\r
- s.timefield = optarg;\r
- break;\r
- case 'e':\r
- s.extension = optarg;\r
- break;\r
- case 'n':\r
- s.notemp=1;\r
- break;\r
- case 't':\r
- s.interval = atoi(optarg);\r
- if (s.interval < 1) {\r
- goto usage;\r
- }\r
- break;\r
- case 's':\r
- s.stream++;\r
- break;\r
- case 'f':\r
- s.flush++;\r
- break;\r
- case 'v':\r
- s.verbose++;\r
- break;\r
- case 'z':\r
- s.compressed++;\r
- break;\r
- case 'h':\r
- s.remote_print=atoi(optarg);\r
- break;\r
- default:\r
- usage:\r
- print_usage_exit("invalid args");\r
- }\r
- }\r
- \r
- if ((s.stream!=0) & (s.compressed!=0 | s.verbose!=0 |\r
- s.timefield!=0 | s.extension!=0)) {\r
- print_usage_exit("illegal argument combination with -s");\r
- }\r
- \r
- if (!s.extension) {\r
- s.extension = (s.compressed) ? ".txt.gz" : ".txt";\r
- }\r
- \r
- argc -= optind;\r
- argv += optind;\r
- if (argc<3) print_usage_exit("must specify hub info and query");\r
- \r
- if (sscanf(argv[0],"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(gshub.port))!= 5 ) {\r
- gslog(LOG_EMERG,"HUB IP NOT DEFINED");\r
- exit(1);\r
- }\r
- gshub.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);\r
- gshub.port=htons(gshub.port);\r
- instance_name=strdup(argv[1]);\r
- if (set_hub(gshub)!=0) {\r
- gslog(LOG_EMERG,"Could not set hub");\r
- exit(1);\r
- }\r
- if (set_instance_name(instance_name)!=0) {\r
- gslog(LOG_EMERG,"Could not set instance name");\r
- exit(1);\r
- }\r
- \r
- if (get_initinstance(gshub,instance_name,&dummyep,1)!=0) {\r
- gslog(LOG_EMERG,"Did not receive signal that GS is initiated\n");\r
- }\r
- \r
- \r
- argc -=2;\r
- argv +=2;\r
- \r
- s.query = argv[0];\r
- \r
- init(argc, argv, &s);\r
- \r
- process_data(&s);\r
- \r
- gslog(LOG_EMERG,"%s::internal error reached unexpected end", me);\r
-}\r
-\r
+/* ------------------------------------------------
+ Copyright 2014 AT&T Intellectual Property
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ ------------------------------------------- */
+#include <app.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <signal.h>
+#include <time.h>
+#include <string.h>
+#include <sys/time.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+
+#include "gsconfig.h"
+#include "gstypes.h"
+#include <schemaparser.h>
+#include <gshub.h>
+
+#define MAXSTRLEN 256
+#define MAXNUMFIELDS 64
+
+//#define GSBZIP2
+
+#ifdef GSBZIP2
+#include "bzlib.h"
+#endif
+
+#ifdef ZLIB
+#include "zlib.h"
+#endif
+
+gs_sp_t me; /* saved copy argv[0] */
+
+struct FTA_state {
+ FTAID fta_id;
+ gs_schemahandle_t schema;
+ gs_sp_t asciischema;
+ gs_int32_t numfields;
+ gs_int32_t timefieldoffset;
+};
+
+struct gsasciiprint_state {
+ // configuration state
+ gs_uint32_t bufsz;
+ gs_int32_t stream;
+ gs_int32_t flush;
+ gs_int32_t parserversion;
+ gs_int32_t compressed;
+ gs_int32_t verbose;
+ gs_int32_t notemp;
+ gs_sp_t timefield;
+ gs_int32_t interval;
+ gs_int32_t quitcnt;
+ gs_int32_t quittime;
+ gs_int32_t numfields;
+ gs_sp_t extension;
+ gs_sp_t query;
+ gs_int32_t remote_print;
+
+
+ // runtime state
+ struct FTA_state fs;
+};
+
+
+void hand(int iv) {
+ ftaapp_exit();
+ gslog(LOG_NOTICE, "exiting via signal handler %d...\n", iv);
+ exit(0);
+}
+
+void timeouthand(int iv) {
+ ftaapp_exit();
+ // if (s->verbose!=0) fprintf(stderr, "exiting because of timeout...\n");
+ exit(0);
+}
+
+
+static void print_usage_exit(gs_sp_t reason) {
+ fprintf(stderr,
+ "%s::error: %s\n"
+ "%s::usage: %s -r <int> -v -f -s -z -c <int> -q <int> -b <field> -t <int> -e <string> <gshub-hostname>:<gshub-port> <gsinstance_name> <query_name> <parameters>\n"
+ "\t-v makes output verbose\n"
+ "\t-f flushes each tuple individually\n"
+ "\t-h <int> print within the hfta int identifies how many output streams based on the SubInterface field\n"
+ "\t-r sets the ringbuffer size default is 8MB\n"
+ "\t-s uses %s in streaming mode cannot be used with -v -z -b -t -e\n"
+ "\t-c <int> indicates that %s should terminate after <int> tuples.\n"
+ "\t-q <int> indicates that %s should terminate after <int> seconds.\n"
+ "\t-z the output is compressed with gzip \n"
+ "\t-b <field> identifies the field which is increasing and is used to\n"
+ "\t\tbin output files. Field has to be of type uint.\n"
+ "\t\tThe default is the UNIX system time\n"
+ "\t-t <int> specifies the number the <field> specified with -b\n"
+ "\t\thas to increase before a new output file is used\n"
+ "\t\tthe default is 60 for 1 minute if the default UNIX time is used\n"
+ "\t-e <string> identifies the file string extension used for the \n"
+ "\t\toutput file. The output file always starts with the current value\n"
+ "\t\tof the b field (or UNIX time) the default extension is .txt \n"
+ "\t\twithout the -z option and .txt.gz with the -z option\n"
+ "\t<query_name> specifies the query which will be\n"
+ "\t\t instanciated.\n"
+ "\t<parameters> sets the parameters for the query\n"
+ , me, reason, me, me, me,me,me);
+ exit(1);
+}
+
+
+
+
+static void init(gs_int32_t argc, gs_sp_t argv[], struct gsasciiprint_state *s) {
+ void *pblk;
+ gs_int32_t x, y, schema, pblklen, lcv;
+ gs_int32_t n_actual_param;
+ gs_int32_t n_expected_param;
+ gs_sp_t c;
+ gs_int8_t name[1024];
+
+ sprintf(name,"gsgdatprint:%s",argv[0]);
+
+ gsopenlog(name);
+
+
+ if (s->verbose!=0) gslog(LOG_DEBUG,"Initializing gscp\n");
+ if (ftaapp_init(s->bufsz)!=0) {
+ gslog(LOG_EMERG,"%s::error:could not initialize gscp\n", me);
+ exit(1);
+ }
+
+ signal(SIGTERM, hand);
+ signal(SIGINT, hand);
+ signal(SIGPIPE, hand);
+
+ if (s->verbose!=0) gslog(LOG_DEBUG,"Initializing FTAs\n");
+
+ schema = ftaapp_get_fta_schema_by_name(argv[0]);
+ if (schema < 0) {
+ gslog(LOG_EMERG,"%s::error:can't get fta '%s' schema\n", me ,argv[0]);
+ exit(1);
+ }
+ n_expected_param = ftaschema_parameter_len(schema);
+ if (n_expected_param == 0) {
+ pblk = 0;
+ pblklen = 0;
+ if (s->verbose) gslog(LOG_DEBUG,"[query does not have any params]\n");
+ } else {
+ n_actual_param = argc-1;
+ if(n_actual_param < n_expected_param){
+ gslog(LOG_EMERG,"Error, %d query parameters expected, %d provided.\n",n_expected_param, n_actual_param);
+ exit(1);
+ }
+ /* parse the params */
+ for (lcv = 1 ; lcv < argc ; lcv++) {
+ char *k, *e;
+ int rv;
+ k = argv[lcv];
+ e = k;
+ while (*e && *e != '=') e++;
+ if (*e == 0) {
+ gslog(LOG_EMERG,"param parse error '%s' (fmt 'key=val')\n",
+ argv[lcv]);
+ exit(1);
+ }
+ *e = 0;
+ rv = ftaschema_setparam_by_name(schema, k, e+1, strlen(e+1));
+ *e = '=';
+ if (rv < 0) {
+ gslog(LOG_EMERG,"param setparam error '%s' (fmt 'key=val')\n",
+ argv[lcv]);
+ exit(1);
+ }
+ }
+ if (ftaschema_create_param_block(schema, &pblk, &pblklen) < 0) {
+ gslog(LOG_EMERG, "ftaschema_create_param_block failed!\n");
+ exit(1);
+ }
+ }
+ ftaschema_free(schema); /* XXXCDC */
+
+
+ if (s->remote_print>=0) {
+ s->fs.fta_id=ftaapp_add_fta_print(s->query,pblk?0:1,pblk?0:1,0,pblklen,pblk,"./data",s->extension,s->timefield,"SubInterface",s->interval,s->remote_print);
+ if (s->fs.fta_id.streamid==0){
+ gslog(LOG_EMERG,"%s::error:could not initialize fta_print %s\n",
+ me,s->query);
+ exit(1);
+ }
+ again:
+ // wait forever
+ sleep(60);
+ goto again;
+ }
+
+
+ s->fs.fta_id=ftaapp_add_fta(s->query,pblk?0:1,pblk?0:1,0,pblklen,pblk);
+ if (s->fs.fta_id.streamid==0){
+ gslog(LOG_EMERG,"%s::error:could not initialize fta %s\n",
+ me,s->query);
+ exit(1);
+ }
+ /* XXXCDC: pblk is malloc'd, should we free it? */
+
+ if ((c=ftaapp_get_fta_ascii_schema_by_name(s->query))==0){
+ gslog(LOG_EMERG,"%s::error:could not get ascii schema for %s\n",
+ me,s->query);
+ exit(1);
+ }
+
+ //ftaapp_get_fta_ascii_schema_by_name uses static buffer so make a copy
+ s->fs.asciischema=strdup(c);
+
+
+ // Set parser version here
+ s->parserversion=get_schemaparser_version();
+
+ if ((s->fs.schema=ftaapp_get_fta_schema(s->fs.fta_id))<0) {
+ gslog(LOG_EMERG,"%s::error:could not get schema for query\n",
+ me,s->query);
+ exit(1);
+ }
+
+ // Use all available fields
+ if ((s->fs.numfields=ftaschema_tuple_len(s->fs.schema))<0) {
+ gslog(LOG_EMERG,"%s::error:could not get number of fields for query %s\n",
+ me,s->query);
+ exit(1);
+ }
+ if (s->timefield!=0) {
+ if ((s->fs.timefieldoffset=ftaschema_get_field_offset_by_name(
+ s->fs.schema,s->timefield))<0) {
+ gslog(LOG_EMERG,"%s::error:could not get "
+ "offset for timefield %s in query %s\n",
+ me,s->timefield,s->query);
+ exit(1);
+ }
+
+ if (ftaschema_get_field_type_by_name(
+ s->fs.schema,s->timefield)!=UINT_TYPE) {
+ gslog(LOG_EMERG,"%s::error: illegal type for timefield "
+ "%s in query %s UINT expected\n",
+ me,s->timefield,s->query);
+ exit(1);
+ }
+ }
+}
+
+
+static void process_data(struct gsasciiprint_state *s)
+{
+ gs_int32_t x;
+ gs_int32_t y;
+ gs_int32_t z;
+ gs_int32_t tb;
+ gs_uint32_t nsz;
+ FTAID rfta_id;
+ gs_uint32_t rsize;
+ gs_int8_t rbuf[2*MAXTUPLESZ];
+ gs_int32_t problem;
+
+ gs_int32_t ctb=0;
+ gs_int8_t fname[1024];
+ gs_int8_t tmpname[1024];
+ gs_int8_t command[1024];
+ gs_int8_t topb[1024];
+ gs_int32_t rcnt=0;
+ gs_retval_t code;
+#ifdef GSBZIP2
+ int bzerror;
+ BZFILE * b;
+ int abandon=0;
+ unsigned int bytes_in;
+ unsigned int bytes_out;
+#endif
+#ifdef ZLIB
+ gzFile of=0;
+#else
+ FILE * of=0;
+#endif
+
+
+ tmpname[0]=0;
+
+ if (s->verbose!=0) gslog(LOG_INFO,"Getting Data for %s",s->query);
+
+ sprintf(&topb[0],"GDAT\nVERSION:%u\nSCHEMALENGTH:%lu\n",
+ s->parserversion,strlen(s->fs.asciischema)+1);
+ if (s->stream!=0) {
+ of=stdout;
+ // need to get ASCII version of schema
+ fwrite(&topb[0],strlen(topb),1,of);
+ fwrite(s->fs.asciischema,strlen(s->fs.asciischema)+1,1,of);
+ }
+
+ if (s->quittime !=0 ) {
+ signal(SIGALRM,timeouthand);
+ alarm(s->quittime);
+ }
+
+
+ while((code=ftaapp_get_tuple(&rfta_id,&rsize,rbuf,2*MAXTUPLESZ,0))>=0) {
+ rcnt++;
+ if ((s->notemp==1) && (code==2)) continue;
+ if (((s->quitcnt>0) && (rcnt>s->quitcnt))||ftaschema_is_eof_tuple(s->fs.schema, rbuf)) {
+ if (s->verbose!=0)
+ gslog(LOG_EMERG, "exiting reached tuple limit or all data has been proccessed\n");
+ if (of!=0) {
+ fclose(of);
+
+ if (s->compressed) {
+ system(command);
+ }
+ rename(tmpname,fname);
+ }
+ exit(0);
+ }
+ if (((code==0) || (code==2))&&(s->fs.fta_id.streamid==rfta_id.streamid)) {
+ if (s->stream==0) {
+ if (s->timefield!=0) {
+ tb=fta_unpack_uint(rbuf,
+ rsize,
+ s->fs.timefieldoffset,
+ &problem);
+ } else {
+ tb=time(0);
+ }
+ if ((ctb+s->interval)<=tb) {
+ gsstats();
+ if (of!=0) {
+#ifdef GSBZIP2
+ BZ2_bzWriteClose(&bzerror, b, abandon, &bytes_in, &bytes_out );
+ if (bzerror!=BZ_OK) {
+ gslog(LOG_EMERG,"Could not bz close file .. EXITING\n");
+ exit(0);
+ }
+#endif
+#ifdef ZLIB
+ gzclose(of);
+#else
+ fclose(of);
+#endif
+#ifndef GSBZIP2
+ if (s->compressed) {
+ system(command);
+ }
+#endif
+ rename(tmpname,fname);
+ }
+ while((ctb+s->interval)<=tb) {
+ if (ctb==0) {
+ ctb=(tb/s->interval)*s->interval;
+ } else {
+ ctb=ctb+s->interval;
+ }
+ }
+#ifdef ZLIB
+ sprintf(tmpname,"%u%s.gz.tmp",ctb,s->extension);
+ sprintf(fname,"%u%s.gz",ctb,s->extension);
+#else
+ sprintf(tmpname,"%u%s.tmp",ctb,s->extension);
+ sprintf(fname,"%u%s",ctb,s->extension);
+#endif
+#ifndef GSBZIP2
+ if (s->compressed) {
+ sprintf(command,"gzip -S .tmpgz %s ; mv %s.tmpgz %s",tmpname,tmpname,tmpname);
+ }
+#endif
+#ifdef ZLIB
+ if ((of=gzopen(tmpname,"wb"))==0) {
+ gslog(LOG_EMERG,"Could not open file \"%s\".. EXITING\n",
+ tmpname);
+ exit(0);
+ }
+#else
+ if ((of=fopen(tmpname,"w"))==0) {
+ gslog(LOG_EMERG,"Could not open file \"%s\".. EXITING\n",
+ tmpname);
+ exit(0);
+ }
+#endif
+#ifdef GSBZIP2
+ if (s->compressed) {
+ b=BZ2_bzWriteOpen(&bzerror,of,5,0,30);
+ if (bzerror!=BZ_OK) {
+ gslog(LOG_EMERG,"Could not bz open file \"%s\".. EXITING\n",
+ tmpname);
+ exit(0);
+ }
+ }
+#endif
+
+ if (s->compressed) {
+#ifdef GSBZIP2
+ BZ2_bzWrite ( &bzerror, b, &topb[0],strlen(topb));
+ BZ2_bzWrite ( &bzerror, b, s->fs.asciischema,strlen(s->fs.asciischema)+1);
+#endif
+ } else {
+ // need to get ASCII version of schema
+#ifdef ZLIB
+ gzwrite(of,&topb[0],strlen(topb));
+ gzwrite(of,s->fs.asciischema,strlen(s->fs.asciischema)+1);
+#else
+ fwrite(&topb[0],strlen(topb),1,of);
+ fwrite(s->fs.asciischema,strlen(s->fs.asciischema)+1,1,of);
+#endif
+ }
+ }
+ }
+ if (code==0) {
+ nsz=htonl(rsize);
+ if (s->compressed) {
+#ifdef GSBZIP2
+ BZ2_bzWrite ( &bzerror, b,&nsz,sizeof(gs_uint32_t) );
+ BZ2_bzWrite ( &bzerror, b,rbuf,rsize);
+#endif
+ } else {
+#ifdef ZLIB
+ gzwrite(of,&nsz,sizeof(gs_uint32_t));
+ gzwrite(of,rbuf,rsize);
+ }
+#else
+ if (fwrite(&nsz,sizeof(gs_uint32_t),1,of)!=1) {
+ ftaapp_exit();
+ if (s->verbose!=0)
+ gslog(LOG_EMERG,"Could not write to output\"%s\".. EXITING\n",
+ tmpname);
+ exit(0);
+ }
+
+ if (fwrite(rbuf,rsize,1,of)!=1) {
+ ftaapp_exit();
+ if (s->verbose!=0)
+ gslog(LOG_EMERG,"Could not write to output\"%s\".. EXITING\n",
+ tmpname);
+ exit(0);
+ }
+ if ((s->stream!=0) || (s->flush!=0)) {
+ fflush(of);
+ }
+ }
+#endif
+ }
+ }
+}
+}
+
+
+
+int main(int argc, char** argv) {
+
+ struct gsasciiprint_state s;
+ gs_retval_t ch;
+ endpoint gshub;
+ endpoint dummyep;
+ gs_uint32_t tip1,tip2,tip3,tip4;
+ gs_sp_t instance_name;
+
+ me = argv[0];
+
+ /* initialize host library and the sgroup */
+
+ if (argc<=1) {
+ print_usage_exit("Not enough arguments");
+ }
+
+ /* parse args */
+ bzero(&s, sizeof(s));
+ s.interval = 60; /* default */
+ s.quittime=0;
+ s.quitcnt=-1;
+ s.remote_print=-1;
+ s.bufsz=8*1024*1024;
+
+ while ((ch = getopt(argc, argv, "nr:b:e:t:c:q:sfvzmh:")) != -1) {
+ switch(ch) {
+ case 'r':
+ s.bufsz=atoi(optarg);
+ break;
+ case 'c':
+ s.quitcnt = atoi(optarg);
+ break;
+ case 'q':
+ s.quittime = atoi(optarg);
+ break;
+ case 'b':
+ s.timefield = optarg;
+ break;
+ case 'e':
+ s.extension = optarg;
+ break;
+ case 'n':
+ s.notemp=1;
+ break;
+ case 't':
+ s.interval = atoi(optarg);
+ if (s.interval < 1) {
+ goto usage;
+ }
+ break;
+ case 's':
+ s.stream++;
+ break;
+ case 'f':
+ s.flush++;
+ break;
+ case 'v':
+ s.verbose++;
+ break;
+ case 'z':
+ s.compressed++;
+ break;
+ case 'h':
+ s.remote_print=atoi(optarg);
+ break;
+ default:
+ usage:
+ print_usage_exit("invalid args");
+ }
+ }
+
+ if ((s.stream!=0) & (s.compressed!=0 | s.verbose!=0 |
+ s.timefield!=0 | s.extension!=0)) {
+ print_usage_exit("illegal argument combination with -s");
+ }
+
+ if (!s.extension) {
+ s.extension = (s.compressed) ? ".txt.gz" : ".txt";
+ }
+
+ argc -= optind;
+ argv += optind;
+ if (argc<3) print_usage_exit("must specify hub info and query");
+
+ if (sscanf(argv[0],"%u.%u.%u.%u:%hu",&tip1,&tip2,&tip3,&tip4,&(gshub.port))!= 5 ) {
+ gslog(LOG_EMERG,"HUB IP NOT DEFINED");
+ exit(1);
+ }
+ gshub.ip=htonl(tip1<<24|tip2<<16|tip3<<8|tip4);
+ gshub.port=htons(gshub.port);
+ instance_name=strdup(argv[1]);
+ if (set_hub(gshub)!=0) {
+ gslog(LOG_EMERG,"Could not set hub");
+ exit(1);
+ }
+ if (set_instance_name(instance_name)!=0) {
+ gslog(LOG_EMERG,"Could not set instance name");
+ exit(1);
+ }
+
+ if (get_initinstance(gshub,instance_name,&dummyep,1)!=0) {
+ gslog(LOG_EMERG,"Did not receive signal that GS is initiated\n");
+ }
+
+
+ argc -=2;
+ argv +=2;
+
+ s.query = argv[0];
+
+ init(argc, argv, &s);
+
+ process_data(&s);
+
+ gslog(LOG_EMERG,"%s::internal error reached unexpected end", me);
+}
+
Code Review / com / gs-lite.git / blobdiff