Mirror/zeek
1
0
Fork 0
mirror of https://github.com/zeek/zeek.git synced 2025-10-05 08:08:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Input framework merge in progress.

Browse source
This commit is contained in:
Robin Sommer 2012-05-25 15:14:25 -07:00
parent 658b188dff
commit b37f9e38f6
16 changed files with 1063 additions and 1050 deletions
Download patch file Download diff file Expand all files Collapse all files

31
scripts/base/frameworks/input/main.bro
View file

@ -18,7 +18,7 @@ export {
## For `READER_ASCII`, this is the filename.
source: string;
## Reader to use for this steam
## Reader to use for this stream
reader: Reader &default=default_reader;
## Read mode to use for this stream
@ -27,27 +27,31 @@ export {
## Descriptive name. Used to remove a stream at a later time
name: string;
## Special definitions for tables
# Special definitions for tables
## Table which will contain the data read by the input framework
## Table which will receive the data read by the input framework
destination: any;
## Record that defines the values used as the index of the table
idx: any;
## Record that defines the values used as the values of the table
## Record that defines the values used as the elements of the table
## If val is undefined, destination has to be a set.
val: any &optional;
## Defines if the value of the table is a record (default), or a single value.
## Val can only contain one element when this is set to false.
## Defines if the value of the table is a record (default), or a single value. Val
## can only contain one element when this is set to false.
want_record: bool &default=T;
## The event that is raised each time a value is added to, changed in or removed from the table.
## The event will receive an Input::Event enum as the first argument, the idx record as the second argument
## and the value (record) as the third argument.
## The event that is raised each time a value is added to, changed in or removed
## from the table. The event will receive an Input::Event enum as the first
## argument, the idx record as the second argument and the value (record) as the
## third argument.
ev: any &optional; # event containing idx, val as values.
## Predicate function, that can decide if an insertion, update or removal should really be executed.
## Parameters are the same as for the event. If true is returned, the update is performed. If false
## is returned, it is skipped
## Predicate function that can decide if an insertion, update or removal should
## really be executed. Parameters are the same as for the event. If true is
## returned, the update is performed. If false is returned, it is skipped.
pred: function(typ: Input::Event, left: any, right: any): bool &optional;
};
@ -68,10 +72,11 @@ export {
## Descriptive name. Used to remove a stream at a later time
name: string;
## Special definitions for events
# Special definitions for events
## Record describing the fields to be retrieved from the source input.
fields: any;
## If want_record if false (default), the event receives each value in fields as a seperate argument.
## If it is set to true, the event receives all fields in a signle record value.
want_record: bool &default=F;

2
scripts/base/frameworks/input/readers/ascii.bro
View file

@ -1,4 +1,6 @@
##! Interface for the ascii input reader.
##!
##! The defaults are set to match Bro's ASCII output.
module InputAscii;

399
src/input/Manager.cc
View file

@ -23,24 +23,38 @@ using namespace input;
using threading::Value;
using threading::Field;
struct ReaderDefinition {
bro_int_t type; // The reader type.
const char *name; // Descriptive name for error messages.
bool (*init)(); // Optional one-time initializing function.
ReaderBackend* (*factory)(ReaderFrontend* frontend); // Factory function for creating instances.
};
ReaderDefinition input_readers[] = {
{ BifEnum::Input::READER_ASCII, "Ascii", 0, reader::Ascii::Instantiate },
{ BifEnum::Input::READER_RAW, "Raw", 0, reader::Raw::Instantiate },
{ BifEnum::Input::READER_BENCHMARK, "Benchmark", 0, reader::Benchmark::Instantiate },
// End marker
{ BifEnum::Input::READER_DEFAULT, "None", 0, (ReaderBackend* (*)(ReaderFrontend* frontend))0 }
};
/**
* InputHashes are used as Dictionaries to store the value and index hashes for all
* lines currently stored in a table. Index hash is stored as HashKey*, because it is
* thrown into other Bro functions that need the complex structure of it.
* For everything we do (with values), we just take the hash_t value and compare it
* directly with ==
* InputHashes are used as Dictionaries to store the value and index hashes
* for all lines currently stored in a table. Index hash is stored as
* HashKey*, because it is thrown into other Bro functions that need the
* complex structure of it. For everything we do (with values), we just take
* the hash_t value and compare it directly with "=="
*/
struct InputHash
{
struct InputHash {
hash_t valhash;
HashKey* idxkey;
~InputHash();
};
};
InputHash::~InputHash()
{
if ( idxkey )
delete idxkey;
delete idxkey;
}
static void input_hash_delete_func(void* val)
@ -52,7 +66,7 @@ static void input_hash_delete_func(void* val)
declare(PDict, InputHash);
/**
* Base stuff that every stream can do
* Base stuff that every stream can do.
*/
class Manager::Stream {
public:
@ -75,20 +89,21 @@ public:
Manager::Stream::Stream()
{
type = 0;
reader = 0;
description = 0;
type = 0;
reader = 0;
description = 0;
removed = false;
}
Manager::Stream::~Stream()
{
if ( type )
if ( type )
Unref(type);
if ( description )
if ( description )
Unref(description);
if ( reader )
if ( reader )
delete(reader);
}
@ -129,7 +144,7 @@ public:
Manager::TableStream::TableStream() : Manager::Stream::Stream()
{
stream_type = TABLE_FILTER;
stream_type = TABLE_STREAM;
tab = 0;
itype = 0;
@ -144,7 +159,7 @@ Manager::TableStream::TableStream() : Manager::Stream::Stream()
Manager::EventStream::EventStream() : Manager::Stream::Stream()
{
fields = 0;
stream_type = EVENT_FILTER;
stream_type = EVENT_STREAM;
}
Manager::EventStream::~EventStream()
@ -157,8 +172,10 @@ Manager::TableStream::~TableStream()
{
if ( tab )
Unref(tab);
if ( itype )
Unref(itype);
if ( rtype ) // can be 0 for sets
Unref(rtype);
@ -175,22 +192,6 @@ Manager::TableStream::~TableStream()
}
}
struct ReaderDefinition {
bro_int_t type; // the type
const char *name; // descriptive name for error messages
bool (*init)(); // optional one-time inifializing function
ReaderBackend* (*factory)(ReaderFrontend* frontend); // factory function for creating instances
};
ReaderDefinition input_readers[] = {
{ BifEnum::Input::READER_ASCII, "Ascii", 0, reader::Ascii::Instantiate },
{ BifEnum::Input::READER_RAW, "Raw", 0, reader::Raw::Instantiate },
{ BifEnum::Input::READER_BENCHMARK, "Benchmark", 0, reader::Benchmark::Instantiate },
// End marker
{ BifEnum::Input::READER_DEFAULT, "None", 0, (ReaderBackend* (*)(ReaderFrontend* frontend))0 }
};
Manager::Manager()
{
}
@ -233,6 +234,7 @@ ReaderBackend* Manager::CreateBackend(ReaderFrontend* frontend, bro_int_t type)
//clear it to be not called again
ir->init = 0;
}
else {
// ohok. init failed, kill factory for all eternity
ir->factory = 0;
@ -242,7 +244,7 @@ ReaderBackend* Manager::CreateBackend(ReaderFrontend* frontend, bro_int_t type)
}
if ( !ir->factory )
if ( ! ir->factory )
// no factory?
return 0;
@ -259,7 +261,7 @@ ReaderBackend* Manager::CreateBackend(ReaderFrontend* frontend, bro_int_t type)
return backend;
}
// create a new input reader object to be used at whomevers leisure lateron.
// Create a new input reader object to be used at whomevers leisure lateron.
bool Manager::CreateStream(Stream* info, RecordVal* description)
{
ReaderDefinition* ir = input_readers;
@ -276,22 +278,20 @@ bool Manager::CreateStream(Stream* info, RecordVal* description)
string name = name_val->AsString()->CheckString();
Unref(name_val);
Stream *i = FindStream(name);
if ( i != 0 )
{
Stream *i = FindStream(name);
if ( i != 0 )
{
reporter->Error("Trying create already existing input stream %s",
name.c_str());
return false;
}
reporter->Error("Trying create already existing input stream %s",
name.c_str());
return false;
}
EnumVal* reader = description->LookupWithDefault(rtype->FieldOffset("reader"))->AsEnumVal();
ReaderFrontend* reader_obj = new ReaderFrontend(reader->InternalInt());
assert(reader_obj);
ReaderFrontend* reader_obj = new ReaderFrontend(reader->InternalInt());
assert(reader_obj);
// get the source...
// get the source ...
Val* sourceval = description->LookupWithDefault(rtype->FieldOffset("source"));
assert ( sourceval != 0 );
const BroString* bsource = sourceval->AsString();
@ -313,12 +313,10 @@ bool Manager::CreateStream(Stream* info, RecordVal* description)
name.c_str());
return true;
}
bool Manager::CreateEventStream(RecordVal* fval)
{
RecordType* rtype = fval->Type()->AsRecordType();
if ( ! same_type(rtype, BifType::Record::Input::EventDescription, 0) )
{
@ -345,73 +343,72 @@ bool Manager::CreateEventStream(RecordVal* fval)
Func* event = event_val->AsFunc();
Unref(event_val);
FuncType* etype = event->FType()->AsFuncType();
if ( ! etype->IsEvent() )
{
FuncType* etype = event->FType()->AsFuncType();
reporter->Error("stream event is a function, not an event");
return false;
}
if ( ! etype->IsEvent() )
const type_list* args = etype->ArgTypes()->Types();
if ( args->length() < 2 )
{
reporter->Error("event takes not enough arguments");
return false;
}
if ( ! same_type((*args)[1], BifType::Enum::Input::Event, 0) )
{
reporter->Error("events second attribute must be of type Input::Event");
return false;
}
if ( ! same_type((*args)[0], BifType::Record::Input::EventDescription, 0) )
{
reporter->Error("events first attribute must be of type Input::EventDescription");
return false;
}
if ( want_record->InternalInt() == 0 )
{
if ( args->length() != fields->NumFields() + 2 )
{
reporter->Error("stream event is a function, not an event");
reporter->Error("event has wrong number of arguments");
return false;
}
const type_list* args = etype->ArgTypes()->Types();
if ( args->length() < 2 )
for ( int i = 0; i < fields->NumFields(); i++ )
{
reporter->Error("event takes not enough arguments");
return false;
}
if ( ! same_type((*args)[1], BifType::Enum::Input::Event, 0) )
{
reporter->Error("events second attribute must be of type Input::Event");
return false;
}
if ( ! same_type((*args)[0], BifType::Record::Input::EventDescription, 0) )
{
reporter->Error("events first attribute must be of type Input::EventDescription");
return false;
}
if ( want_record->InternalInt() == 0 )
{
if ( args->length() != fields->NumFields() + 2 )
{
reporter->Error("event has wrong number of arguments");
return false;
}
for ( int i = 0; i < fields->NumFields(); i++ )
{
if ( !same_type((*args)[i+2], fields->FieldType(i) ) )
{
reporter->Error("Incompatible type for event");
return false;
}
}
}
else if ( want_record->InternalInt() == 1 )
{
if ( args->length() != 3 )
{
reporter->Error("event has wrong number of arguments");
return false;
}
if ( !same_type((*args)[2], fields ) )
if ( !same_type((*args)[i+2], fields->FieldType(i) ) )
{
reporter->Error("Incompatible type for event");
return false;
}
}
else
assert(false);
}
else if ( want_record->InternalInt() == 1 )
{
if ( args->length() != 3 )
{
reporter->Error("event has wrong number of arguments");
return false;
}
if ( !same_type((*args)[2], fields ) )
{
reporter->Error("Incompatible type for event");
return false;
}
}
else
assert(false);
vector<Field*> fieldsV; // vector, because UnrollRecordType needs it
@ -478,30 +475,28 @@ bool Manager::CreateTableStream(RecordVal* fval)
TableVal *dst = fval->LookupWithDefault(rtype->FieldOffset("destination"))->AsTableVal();
// check if index fields match table description
int num = idx->NumFields();
const type_list* tl = dst->Type()->AsTableType()->IndexTypes();
loop_over_list(*tl, j)
{
int num = idx->NumFields();
const type_list* tl = dst->Type()->AsTableType()->IndexTypes();
loop_over_list(*tl, j)
if ( j >= num )
{
if ( j >= num )
{
reporter->Error("Table type has more indexes than index definition");
return false;
}
if ( !same_type(idx->FieldType(j), (*tl)[j]) )
{
reporter->Error("Table type does not match index type");
return false;
}
}
if ( num != j )
{
reporter->Error("Table has less elements than index definition");
reporter->Error("Table type has more indexes than index definition");
return false;
}
if ( ! same_type(idx->FieldType(j), (*tl)[j]) )
{
reporter->Error("Table type does not match index type");
return false;
}
}
if ( num != j )
{
reporter->Error("Table has less elements than index definition");
return false;
}
Val *want_record = fval->LookupWithDefault(rtype->FieldOffset("want_record"));
@ -573,7 +568,7 @@ bool Manager::CreateTableStream(RecordVal* fval)
int valfields = fieldsV.size() - idxfields;
if ( !val )
if ( ! val )
assert(valfields == 0);
if ( status )
@ -582,7 +577,6 @@ bool Manager::CreateTableStream(RecordVal* fval)
return false;
}
Field** fields = new Field*[fieldsV.size()];
for ( unsigned int i = 0; i < fieldsV.size(); i++ )
fields[i] = fieldsV[i];
@ -718,6 +712,7 @@ bool Manager::RemoveStreamContinuation(ReaderFrontend* reader)
readers.erase(reader);
delete(i);
return true;
}
@ -728,7 +723,7 @@ bool Manager::UnrollRecordType(vector<Field*> *fields,
for ( int i = 0; i < rec->NumFields(); i++ )
{
if ( !IsCompatibleType(rec->FieldType(i)) )
if ( ! IsCompatibleType(rec->FieldType(i)) )
{
reporter->Error("Incompatible type \"%s\" in table definition for ReaderFrontend", type_name(rec->FieldType(i)->Tag()));
return false;
@ -744,20 +739,21 @@ bool Manager::UnrollRecordType(vector<Field*> *fields,
}
}
else
{
Field* field = new Field();
field->name = nameprepend + rec->FieldName(i);
field->type = rec->FieldType(i)->Tag();
if ( field->type == TYPE_TABLE )
{
field->subtype = rec->FieldType(i)->AsSetType()->Indices()->PureType()->Tag();
}
else if ( field->type == TYPE_VECTOR )
{
field->subtype = rec->FieldType(i)->AsVectorType()->YieldType()->Tag();
} else if ( field->type == TYPE_PORT &&
rec->FieldDecl(i)->FindAttr(ATTR_TYPE_COLUMN) )
else if ( field->type == TYPE_PORT &&
rec->FieldDecl(i)->FindAttr(ATTR_TYPE_COLUMN) )
{
// we have an annotation for the second column
@ -797,8 +793,7 @@ bool Manager::ForceUpdate(const string &name)
i->reader->Update();
#ifdef DEBUG
DBG_LOG(DBG_INPUT, "Forcing update of stream %s",
name.c_str());
DBG_LOG(DBG_INPUT, "Forcing update of stream %s", name.c_str());
#endif
return true; // update is async :(
@ -814,9 +809,8 @@ Val* Manager::RecordValToIndexVal(RecordVal *r)
int num_fields = type->NumFields();
if ( num_fields == 1 && type->FieldDecl(0)->type->Tag() != TYPE_RECORD )
{
idxval = r->LookupWithDefault(0);
}
else
{
ListVal *l = new ListVal(TYPE_ANY);
@ -837,10 +831,14 @@ Val* Manager::ValueToIndexVal(int num_fields, const RecordType *type, const Valu
int position = 0;
if ( num_fields == 1 && type->FieldType(0)->Tag() != TYPE_RECORD ) {
if ( num_fields == 1 && type->FieldType(0)->Tag() != TYPE_RECORD )
{
idxval = ValueToVal(vals[0], type->FieldType(0));
position = 1;
} else {
}
else
{
ListVal *l = new ListVal(TYPE_ANY);
for ( int j = 0 ; j < type->NumFields(); j++ )
{
@ -871,14 +869,17 @@ void Manager::SendEntry(ReaderFrontend* reader, Value* *vals)
return;
}
int readFields;
if ( i->stream_type == TABLE_FILTER )
int readFields = 0;
if ( i->stream_type == TABLE_STREAM )
readFields = SendEntryTable(i, vals);
else if ( i->stream_type == EVENT_FILTER )
else if ( i->stream_type == EVENT_STREAM )
{
EnumVal *type = new EnumVal(BifEnum::Input::EVENT_NEW, BifType::Enum::Input::Event);
readFields = SendEventStreamEvent(i, type, vals);
}
else
assert(false);
@ -894,7 +895,7 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
assert(i);
assert(i->stream_type == TABLE_FILTER);
assert(i->stream_type == TABLE_STREAM);
TableStream* stream = (TableStream*) i;
HashKey* idxhash = HashValues(stream->num_idx_fields, vals);
@ -909,14 +910,15 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
if ( stream->num_val_fields > 0 )
{
HashKey* valhashkey = HashValues(stream->num_val_fields, vals+stream->num_idx_fields);
if ( valhashkey == 0 ) {
if ( valhashkey == 0 )
{
// empty line. index, but no values.
// hence we also have no hash value...
}
else
{
valhash = valhashkey->Hash();
delete(valhashkey);
valhash = valhashkey->Hash();
delete(valhashkey);
}
}
@ -932,6 +934,7 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
delete idxhash;
return stream->num_val_fields + stream->num_idx_fields;
}
else
{
assert( stream->num_val_fields > 0 );
@ -943,15 +946,17 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
}
}
Val* valval;
RecordVal* predidx = 0;
int position = stream->num_idx_fields;
if ( stream->num_val_fields == 0 )
valval = 0;
else if ( stream->num_val_fields == 1 && !stream->want_record )
valval = ValueToVal(vals[position], stream->rtype->FieldType(0));
else
valval = ValueToRecordVal(vals, stream->rtype, &position);
@ -960,11 +965,8 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
if ( stream->pred )
{
EnumVal* ev;
//Ref(idxval);
int startpos = 0;
//Val* predidx = ListValToRecordVal(idxval->AsListVal(), stream->itype, &startpos);
predidx = ValueToRecordVal(vals, stream->itype, &startpos);
//ValueToRecordVal(vals, stream->itype, &startpos);
if ( updated )
ev = new EnumVal(BifEnum::Input::EVENT_CHANGED, BifType::Enum::Input::Event);
@ -981,7 +983,8 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
{
Unref(predidx);
Unref(valval);
if ( !updated )
if ( ! updated )
{
// throw away. Hence - we quit. And remove the entry from the current dictionary...
// (but why should it be in there? assert this).
@ -990,6 +993,7 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
delete h;
return stream->num_val_fields + stream->num_idx_fields;
}
else
{
// keep old one
@ -998,7 +1002,6 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
return stream->num_val_fields + stream->num_idx_fields;
}
}
}
// now we don't need h anymore - if we are here, the entry is updated and a new h is created.
@ -1010,7 +1013,7 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
Val* idxval;
if ( predidx != 0 )
if ( predidx != 0 )
{
idxval = RecordValToIndexVal(predidx);
// I think there is an unref missing here. But if I insert is, it crashes :)
@ -1026,14 +1029,10 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
oldval = stream->tab->Lookup(idxval, false);
}
//i->tab->Assign(idxval, valval);
assert(idxval);
HashKey* k = stream->tab->ComputeHash(idxval);
if ( !k )
{
if ( ! k )
reporter->InternalError("could not hash");
assert(false);
}
InputHash* ih = new InputHash();
ih->idxkey = new HashKey(k->Key(), k->Size(), k->Hash());
@ -1044,6 +1043,7 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
stream->tab->Assign(idxval, k, valval);
Unref(idxval); // asssign does not consume idxval.
if ( predidx != 0 )
Unref(predidx);
@ -1063,6 +1063,7 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
assert ( oldval != 0 );
SendEvent(stream->event, 4, stream->description->Ref(), ev, predidx, oldval);
}
else
{
ev = new EnumVal(BifEnum::Input::EVENT_NEW, BifType::Enum::Input::Event);
@ -1077,11 +1078,9 @@ int Manager::SendEntryTable(Stream* i, const Value* const *vals)
}
}
return stream->num_val_fields + stream->num_idx_fields;
}
void Manager::EndCurrentSend(ReaderFrontend* reader)
{
Stream *i = FindStream(reader);
@ -1093,14 +1092,13 @@ void Manager::EndCurrentSend(ReaderFrontend* reader)
}
#ifdef DEBUG
DBG_LOG(DBG_INPUT, "Got EndCurrentSend stream %s",
i->name.c_str());
DBG_LOG(DBG_INPUT, "Got EndCurrentSend stream %s", i->name.c_str());
#endif
if ( i->stream_type == EVENT_FILTER ) // nothing to do..
if ( i->stream_type == EVENT_STREAM ) // nothing to do..
return;
assert(i->stream_type == TABLE_FILTER);
assert(i->stream_type == TABLE_STREAM);
TableStream* stream = (TableStream*) i;
// lastdict contains all deleted entries and should be empty apart from that
@ -1108,7 +1106,7 @@ void Manager::EndCurrentSend(ReaderFrontend* reader)
stream->lastDict->MakeRobustCookie(c);
InputHash* ih;
HashKey *lastDictIdxKey;
//while ( ( ih = i->lastDict->NextEntry(c) ) ) {
while ( ( ih = stream->lastDict->NextEntry(lastDictIdxKey, c) ) )
{
ListVal * idx = 0;
@ -1160,6 +1158,7 @@ void Manager::EndCurrentSend(ReaderFrontend* reader)
if ( predidx ) // if we have a stream or an event...
Unref(predidx);
if ( ev )
Unref(ev);
@ -1177,8 +1176,8 @@ void Manager::EndCurrentSend(ReaderFrontend* reader)
stream->currDict->SetDeleteFunc(input_hash_delete_func);
#ifdef DEBUG
DBG_LOG(DBG_INPUT, "EndCurrentSend complete for stream %s, queueing update_finished event",
i->name.c_str());
DBG_LOG(DBG_INPUT, "EndCurrentSend complete for stream %s, queueing update_finished event",
i->name.c_str());
#endif
// Send event that the current update is indeed finished.
@ -1198,14 +1197,17 @@ void Manager::Put(ReaderFrontend* reader, Value* *vals)
return;
}
int readFields;
if ( i->stream_type == TABLE_FILTER )
int readFields = 0;
if ( i->stream_type == TABLE_STREAM )
readFields = PutTable(i, vals);
else if ( i->stream_type == EVENT_FILTER )
else if ( i->stream_type == EVENT_STREAM )
{
EnumVal *type = new EnumVal(BifEnum::Input::EVENT_NEW, BifType::Enum::Input::Event);
readFields = SendEventStreamEvent(i, type, vals);
}
else
assert(false);
@ -1219,7 +1221,7 @@ int Manager::SendEventStreamEvent(Stream* i, EnumVal* type, const Value* const *
{
assert(i);
assert(i->stream_type == EVENT_FILTER);
assert(i->stream_type == EVENT_STREAM);
EventStream* stream = (EventStream*) i;
Val *val;
@ -1227,29 +1229,33 @@ int Manager::SendEventStreamEvent(Stream* i, EnumVal* type, const Value* const *
Ref(stream->description);
out_vals.push_back(stream->description);
// no tracking, send everything with a new event...
//out_vals.push_back(new EnumVal(BifEnum::Input::EVENT_NEW, BifType::Enum::Input::Event));
out_vals.push_back(type);
int position = 0;
if ( stream->want_record )
{
RecordVal * r = ValueToRecordVal(vals, stream->fields, &position);
out_vals.push_back(r);
}
else
{
else
{
for ( int j = 0; j < stream->fields->NumFields(); j++)
{
Val* val = 0;
if ( stream->fields->FieldType(j)->Tag() == TYPE_RECORD )
val = ValueToRecordVal(vals,
stream->fields->FieldType(j)->AsRecordType(),
&position);
else
{
val = ValueToVal(vals[position], stream->fields->FieldType(j));
position++;
}
out_vals.push_back(val);
}
}
@ -1257,24 +1263,26 @@ int Manager::SendEventStreamEvent(Stream* i, EnumVal* type, const Value* const *
SendEvent(stream->event, out_vals);
return stream->fields->NumFields();
}
int Manager::PutTable(Stream* i, const Value* const *vals)
{
assert(i);
assert(i->stream_type == TABLE_FILTER);
assert(i->stream_type == TABLE_STREAM);
TableStream* stream = (TableStream*) i;
Val* idxval = ValueToIndexVal(stream->num_idx_fields, stream->itype, vals);
Val* valval;
int position = stream->num_idx_fields;
if ( stream->num_val_fields == 0 )
valval = 0;
else if ( stream->num_val_fields == 1 && stream->want_record == 0 )
valval = ValueToVal(vals[position], stream->rtype->FieldType(0));
else
valval = ValueToRecordVal(vals, stream->rtype, &position);
@ -1364,10 +1372,10 @@ int Manager::PutTable(Stream* i, const Value* const *vals)
}
}
else // no predicates or other stuff
stream->tab->Assign(idxval, valval);
return stream->num_idx_fields + stream->num_val_fields;
}
@ -1386,7 +1394,7 @@ void Manager::Clear(ReaderFrontend* reader)
i->name.c_str());
#endif
assert(i->stream_type == TABLE_FILTER);
assert(i->stream_type == TABLE_STREAM);
TableStream* stream = (TableStream*) i;
stream->tab->RemoveAll();
@ -1405,7 +1413,7 @@ bool Manager::Delete(ReaderFrontend* reader, Value* *vals)
bool success = false;
int readVals = 0;
if ( i->stream_type == TABLE_FILTER )
if ( i->stream_type == TABLE_STREAM )
{
TableStream* stream = (TableStream*) i;
Val* idxval = ValueToIndexVal(stream->num_idx_fields, stream->itype, vals);
@ -1451,19 +1459,21 @@ bool Manager::Delete(ReaderFrontend* reader, Value* *vals)
{
Val* retptr = stream->tab->Delete(idxval);
success = ( retptr != 0 );
if ( !success )
if ( ! success )
reporter->Error("Internal error while deleting values from input table");
else
Unref(retptr);
}
}
else if ( i->stream_type == EVENT_FILTER )
else if ( i->stream_type == EVENT_STREAM )
{
EnumVal *type = new EnumVal(BifEnum::Input::EVENT_REMOVED, BifType::Enum::Input::Event);
readVals = SendEventStreamEvent(i, type, vals);
success = true;
}
else
{
assert(false);
@ -1605,7 +1615,6 @@ RecordVal* Manager::ValueToRecordVal(const Value* const *vals,
RecordVal* rec = new RecordVal(request_type->AsRecordType());
for ( int i = 0; i < request_type->NumFields(); i++ )
{
Val* fieldVal = 0;
if ( request_type->FieldType(i)->Tag() == TYPE_RECORD )
fieldVal = ValueToRecordVal(vals, request_type->FieldType(i)->AsRecordType(), position);
@ -1621,8 +1630,8 @@ RecordVal* Manager::ValueToRecordVal(const Value* const *vals,
return rec;
}
// Count the length of the values
// used to create a correct length buffer for hashing later
// Count the length of the values used to create a correct length buffer for
// hashing later
int Manager::GetValueLength(const Value* val) {
assert( val->present ); // presence has to be checked elsewhere
int length = 0;
@ -1722,13 +1731,11 @@ int Manager::CopyValue(char *data, const int startpos, const Value* val)
case TYPE_INT:
memcpy(data+startpos, (const void*) &(val->val.int_val), sizeof(val->val.int_val));
return sizeof(val->val.int_val);
break;
case TYPE_COUNT:
case TYPE_COUNTER:
memcpy(data+startpos, (const void*) &(val->val.uint_val), sizeof(val->val.uint_val));
return sizeof(val->val.uint_val);
break;
case TYPE_PORT:
{
@ -1740,7 +1747,6 @@ int Manager::CopyValue(char *data, const int startpos, const Value* val)
sizeof(val->val.port_val.proto));
length += sizeof(val->val.port_val.proto);
return length;
break;
}
@ -1750,14 +1756,12 @@ int Manager::CopyValue(char *data, const int startpos, const Value* val)
memcpy(data+startpos, (const void*) &(val->val.double_val),
sizeof(val->val.double_val));
return sizeof(val->val.double_val);
break;
case TYPE_STRING:
case TYPE_ENUM:
{
memcpy(data+startpos, val->val.string_val->c_str(), val->val.string_val->length());
return val->val.string_val->size();
break;
}
case TYPE_ADDR:
@ -1768,16 +1772,18 @@ int Manager::CopyValue(char *data, const int startpos, const Value* val)
length = sizeof(val->val.addr_val.in.in4);
memcpy(data + startpos, (const char*) &(val->val.addr_val.in.in4), length);
break;
case IPv6:
length = sizeof(val->val.addr_val.in.in6);
memcpy(data + startpos, (const char*) &(val->val.addr_val.in.in6), length);
break;
default:
assert(false);
}
return length;
}
break;
case TYPE_SUBNET:
{
@ -1788,21 +1794,24 @@ int Manager::CopyValue(char *data, const int startpos, const Value* val)
memcpy(data + startpos,
(const char*) &(val->val.subnet_val.prefix.in.in4), length);
break;
case IPv6:
length = sizeof(val->val.addr_val.in.in6);
memcpy(data + startpos,
(const char*) &(val->val.subnet_val.prefix.in.in4), length);
break;
default:
assert(false);
}
int lengthlength = sizeof(val->val.subnet_val.length);
memcpy(data + startpos + length ,
(const char*) &(val->val.subnet_val.length), lengthlength);
length += lengthlength;
return length;
}
break;
case TYPE_TABLE:
{
@ -1812,7 +1821,6 @@ int Manager::CopyValue(char *data, const int startpos, const Value* val)
length += CopyValue(data, startpos+length, val->val.set_val.vals[i]);
return length;
break;
}
case TYPE_VECTOR:
@ -1823,7 +1831,6 @@ int Manager::CopyValue(char *data, const int startpos, const Value* val)
length += CopyValue(data, startpos+length, val->val.vector_val.vals[i]);
return length;
break;
}
default:
@ -1894,24 +1901,20 @@ Val* Manager::ValueToVal(const Value* val, BroType* request_type)
case TYPE_COUNT:
case TYPE_COUNTER:
return new Val(val->val.uint_val, val->type);
break;
case TYPE_DOUBLE:
case TYPE_TIME:
case TYPE_INTERVAL:
return new Val(val->val.double_val, val->type);
break;
case TYPE_STRING:
{
BroString *s = new BroString(*(val->val.string_val));
return new StringVal(s);
break;
}
case TYPE_PORT:
return new PortVal(val->val.port_val.port, val->val.port_val.proto);
break;
case TYPE_ADDR:
{
@ -1920,12 +1923,15 @@ Val* Manager::ValueToVal(const Value* val, BroType* request_type)
case IPv4:
addr = new IPAddr(val->val.addr_val.in.in4);
break;
case IPv6:
addr = new IPAddr(val->val.addr_val.in.in6);
break;
default:
assert(false);
}
AddrVal* addrval = new AddrVal(*addr);
delete addr;
return addrval;
@ -1938,16 +1944,18 @@ Val* Manager::ValueToVal(const Value* val, BroType* request_type)
case IPv4:
addr = new IPAddr(val->val.subnet_val.prefix.in.in4);
break;
case IPv6:
addr = new IPAddr(val->val.subnet_val.prefix.in.in6);
break;
default:
assert(false);
}
SubNetVal* subnetval = new SubNetVal(*addr, val->val.subnet_val.length);
delete addr;
return subnetval;
break;
}
case TYPE_TABLE:
@ -1967,7 +1975,6 @@ Val* Manager::ValueToVal(const Value* val, BroType* request_type)
Unref(s);
return t;
break;
}
case TYPE_VECTOR:
@ -1981,7 +1988,6 @@ Val* Manager::ValueToVal(const Value* val, BroType* request_type)
Unref(vt);
return v;
break;
}
case TYPE_ENUM: {
@ -1995,7 +2001,6 @@ Val* Manager::ValueToVal(const Value* val, BroType* request_type)
module.c_str(), var.c_str());
return new EnumVal(index, request_type->Ref()->AsEnumType() );
break;
}

100
src/input/Manager.h
View file

@ -1,6 +1,6 @@
// See the file "COPYING" in the main distribution directory for copyright.
//
// Class for managing input streams
// Class for managing input streams.
#ifndef INPUT_MANAGER_H
#define INPUT_MANAGER_H
@ -34,34 +34,36 @@ public:
~Manager();
/**
* Creates a new input stream which will write the data from the data source into
* Creates a new input stream which will write the data from the data
* source into a table.
*
* @param description A record of script type \c Input:StreamDescription.
* @param description A record of script type \c
* Input:StreamDescription.
*
* This method corresponds directly to the internal BiF defined in
* input.bif, which just forwards here.
*/
bool CreateTableStream(RecordVal* description);
bool CreateTableStream(RecordVal* description);
/**
* Creates a new input stream which sends events for read input data.
*
* @param description A record of script type \c Input:StreamDescription.
* @param description A record of script type \c
* Input:StreamDescription.
*
* This method corresponds directly to the internal BiF defined in
* input.bif, which just forwards here.
*/
bool CreateEventStream(RecordVal* description);
bool CreateEventStream(RecordVal* description);
/**
* Force update on a input stream.
* Forces a re-read of the whole input source.
* Usually used, when an input stream is opened in managed mode.
* Otherwise, this can be used to trigger a input source check before a heartbeat message arrives.
* May be ignored by the reader.
* Force update on a input stream. Forces a re-read of the whole
* input source. Usually used when an input stream is opened in
* managed mode. Otherwise, this can be used to trigger a input
* source check before a heartbeat message arrives. May be ignored by
* the reader.
*
* @param id The enum value corresponding the input stream.
* @param id The enum value corresponding the input stream.
*
* This method corresponds directly to the internal BiF defined in
* input.bif, which just forwards here.
@ -69,9 +71,9 @@ public:
bool ForceUpdate(const string &id);
/**
* Deletes an existing input stream
* Deletes an existing input stream.
*
* @param id The enum value corresponding the input stream.
* @param id The enum value corresponding the input stream.
*
* This method corresponds directly to the internal BiF defined in
* input.bif, which just forwards here.
@ -88,31 +90,34 @@ protected:
friend class EndCurrentSendMessage;
friend class ReaderClosedMessage;
// For readers to write to input stream in direct mode (reporting new/deleted values directly)
// Functions take ownership of threading::Value fields
// For readers to write to input stream in direct mode (reporting
// new/deleted values directly). Functions take ownership of
// threading::Value fields.
void Put(ReaderFrontend* reader, threading::Value* *vals);
void Clear(ReaderFrontend* reader);
bool Delete(ReaderFrontend* reader, threading::Value* *vals);
// for readers to write to input stream in indirect mode (manager is monitoring new/deleted values)
// Functions take ownership of threading::Value fields
// For readers to write to input stream in indirect mode (manager is
// monitoring new/deleted values) Functions take ownership of
// threading::Value fields.
void SendEntry(ReaderFrontend* reader, threading::Value* *vals);
void EndCurrentSend(ReaderFrontend* reader);
// Allows readers to directly send Bro events.
// The num_vals and vals must be the same the named event expects.
// Takes ownership of threading::Value fields
// Allows readers to directly send Bro events. The num_vals and vals
// must be the same the named event expects. Takes ownership of
// threading::Value fields.
bool SendEvent(const string& name, const int num_vals, threading::Value* *vals);
// Instantiates a new ReaderBackend of the given type (note that
// doing so creates a new thread!).
ReaderBackend* CreateBackend(ReaderFrontend* frontend, bro_int_t type);
// Functions are called from the ReaderBackend to notify the manager, that a stream has been removed
// or a stream has been closed.
// Used to prevent race conditions where data for a specific stream is still in the queue when the
// RemoveStream directive is executed by the main thread.
// This makes sure all data that has ben queued for a stream is still received.
// Function called from the ReaderBackend to notify the manager that
// a stream has been removed or a stream has been closed. Used to
// prevent race conditions where data for a specific stream is still
// in the queue when the RemoveStream directive is executed by the
// main thread. This makes sure all data that has ben queued for a
// stream is still received.
bool RemoveStreamContinuation(ReaderFrontend* reader);
private:
@ -120,57 +125,64 @@ private:
class TableStream;
class EventStream;
bool CreateStream(Stream*, RecordVal* description);
bool CreateStream(Stream*, RecordVal* description);
// SendEntry implementation for Table stream
// SendEntry implementation for Table stream.
int SendEntryTable(Stream* i, const threading::Value* const *vals);
// Put implementation for Table stream
// Put implementation for Table stream.
int PutTable(Stream* i, const threading::Value* const *vals);
// SendEntry and Put implementation for Event stream
// SendEntry and Put implementation for Event stream.
int SendEventStreamEvent(Stream* i, EnumVal* type, const threading::Value* const *vals);
// Checks is a bro type can be used for data reading. The equivalend in threading cannot be used, because we have support different types
// from the log framework
// Checks that a Bro type can be used for data reading. The
// equivalend in threading cannot be used, because we have support
// different types from the log framework
bool IsCompatibleType(BroType* t, bool atomic_only=false);
// Check if a record is made up of compatible types and return a list of all fields that are in the record in order.
// Recursively unrolls records
// Check if a record is made up of compatible types and return a list
// of all fields that are in the record in order. Recursively unrolls
// records
bool UnrollRecordType(vector<threading::Field*> *fields, const RecordType *rec, const string& nameprepend);
// Send events
void SendEvent(EventHandlerPtr ev, const int numvals, ...);
void SendEvent(EventHandlerPtr ev, list<Val*> events);
// Call predicate function and return result
// Call predicate function and return result.
bool CallPred(Func* pred_func, const int numvals, ...);
// get a hashkey for a set of threading::Values
// Get a hashkey for a set of threading::Values.
HashKey* HashValues(const int num_elements, const threading::Value* const *vals);
// Get the memory used by a specific value
// Get the memory used by a specific value.
int GetValueLength(const threading::Value* val);
// Copies the raw data in a specific threading::Value to position sta
// Copies the raw data in a specific threading::Value to position
// startpos.
int CopyValue(char *data, const int startpos, const threading::Value* val);
// Convert Threading::Value to an internal Bro Type (works also with Records)
// Convert Threading::Value to an internal Bro Type (works also with
// Records).
Val* ValueToVal(const threading::Value* val, BroType* request_type);
// Convert Threading::Value to an internal Bro List type
// Convert Threading::Value to an internal Bro List type.
Val* ValueToIndexVal(int num_fields, const RecordType* type, const threading::Value* const *vals);
// Converts a threading::value to a record type. mostly used by ValueToVal
// Converts a threading::value to a record type. Mostly used by
// ValueToVal.
RecordVal* ValueToRecordVal(const threading::Value* const *vals, RecordType *request_type, int* position);
Val* RecordValToIndexVal(RecordVal *r);
// Converts a Bro ListVal to a RecordVal given the record type
// Converts a Bro ListVal to a RecordVal given the record type.
RecordVal* ListValToRecordVal(ListVal* list, RecordType *request_type, int* position);
Stream* FindStream(const string &name);
Stream* FindStream(ReaderFrontend* reader);
enum StreamType { TABLE_FILTER, EVENT_FILTER };
enum StreamType { TABLE_STREAM, EVENT_STREAM };
map<ReaderFrontend*, Stream*> readers;
};

22
src/input/ReaderBackend.cc
View file

@ -64,10 +64,10 @@ public:
{
bool success = input_mgr->SendEvent(name, num_vals, val);
if ( !success )
if ( ! success )
reporter->Error("SendEvent for event %s failed", name.c_str());
return true; // we do not want to die if sendEvent fails because the event did not return.
return true; // We do not want to die if sendEvent fails because the event did not return.
}
private:
@ -136,10 +136,7 @@ public:
ReaderBackend::ReaderBackend(ReaderFrontend* arg_frontend) : MsgThread()
{
buf = 0;
buf_len = 1024;
disabled = true; // disabled will be set correcty in init.
frontend = arg_frontend;
SetName(frontend->Name());
@ -191,7 +188,7 @@ bool ReaderBackend::Init(string arg_source, int mode, const int arg_num_fields,
// disable if DoInit returns error.
int success = DoInit(arg_source, mode, arg_num_fields, arg_fields);
if ( !success )
if ( ! success )
{
Error("Init failed");
DisableFrontend();
@ -214,7 +211,7 @@ void ReaderBackend::Close()
for ( unsigned int i = 0; i < num_fields; i++ )
delete(fields[i]);
delete[] (fields);
delete [] (fields);
fields = 0;
}
}
@ -225,7 +222,7 @@ bool ReaderBackend::Update()
return false;
bool success = DoUpdate();
if ( !success )
if ( ! success )
DisableFrontend();
return success;
@ -233,8 +230,9 @@ bool ReaderBackend::Update()
void ReaderBackend::DisableFrontend()
{
// We also set disabled here, because there still may be other
// messages queued and we will dutifully ignore these from now.
disabled = true;
// we also set disabled here, because there still may be other messages queued and we will dutifully ignore these from now
SendOut(new DisableMessage(frontend));
}
@ -261,7 +259,7 @@ TransportProto ReaderBackend::StringToProto(const string &proto)
}
// more or less verbose copy from IPAddr.cc -- which uses reporter
// More or less verbose copy from IPAddr.cc -- which uses reporter.
Value::addr_t ReaderBackend::StringToAddr(const string &s)
{
Value::addr_t val;
@ -272,7 +270,7 @@ Value::addr_t ReaderBackend::StringToAddr(const string &s)
if ( inet_aton(s.c_str(), &(val.in.in4)) <= 0 )
{
Error(Fmt("Bad addres: %s", s.c_str()));
Error(Fmt("Bad address: %s", s.c_str()));
memset(&val.in.in4.s_addr, 0, sizeof(val.in.in4.s_addr));
}
@ -283,7 +281,7 @@ Value::addr_t ReaderBackend::StringToAddr(const string &s)
val.family = IPv6;
if ( inet_pton(AF_INET6, s.c_str(), val.in.in6.s6_addr) <=0 )
{
Error(Fmt("Bad IP address: %s", s.c_str()));
Error(Fmt("Bad address: %s", s.c_str()));
memset(val.in.in6.s6_addr, 0, sizeof(val.in.in6.s6_addr));
}
}

132
src/input/ReaderBackend.h
View file

@ -12,13 +12,13 @@ namespace input {
class ReaderFrontend;
/**
* Base class for reader implementation. When the input:Manager creates a
* new input stream, it instantiates a ReaderFrontend. That then in turn
* creates a ReaderBackend of the right type. The frontend then forwards
* message over the backend as its methods are called.
* Base class for reader implementation. When the input:Manager creates a new
* input stream, it instantiates a ReaderFrontend. That then in turn creates
* a ReaderBackend of the right type. The frontend then forwards messages
* over the backend as its methods are called.
*
* All of this methods must be called only from the corresponding child
* thread (the constructor is the one exception.)
* All methods must be called only from the corresponding child thread (the
* constructor is the one exception.)
*/
class ReaderBackend : public threading::MsgThread {
public:
@ -27,10 +27,8 @@ public:
*
* @param frontend The frontend reader that created this backend. The
* *only* purpose of this value is to be passed back via messages as
* a argument to callbacks. One must not otherwise access the
* an argument to callbacks. One must not otherwise access the
* frontend, it's running in a different thread.
*
* @param frontend pointer to the reader frontend
*/
ReaderBackend(ReaderFrontend* frontend);
@ -42,39 +40,38 @@ public:
/**
* One-time initialization of the reader to define the input source.
*
* @param arg_source A string left to the interpretation of the reader
* implementation; it corresponds to the value configured on the
* script-level for the input stream.
* @param arg_source A string left to the interpretation of the
* reader implementation; it corresponds to the value configured on
* the script-level for the input stream.
*
* @param num_fields The number of log fields for the stream.
* @param fields An array of size \a num_fields with the input
* fields. The method takes ownership of the array.
*
* @param fields An array of size \a num_fields with the log fields.
* The methods takes ownership of the array.
* @param mode The opening mode for the input source as one of the
* Input::Mode script constants.
*
* @param mode the opening mode for the input source
* @param arg_num_fields Number of fields contained in \a fields.
*
* @param arg_num_fields number of fields contained in \a fields
*
* @param fields the types and names of the fields to be retrieved
* from the input source
* @param fields The types and names of the fields to be retrieved
* from the input source.
*
* @return False if an error occured.
*/
bool Init(string arg_source, int mode, int arg_num_fields, const threading::Field* const* fields);
/**
* Finishes reading from this input stream in a regular fashion. Must not be
* called if an error has been indicated earlier. After calling this,
* no further reading from the stream can be performed
* Finishes reading from this input stream in a regular fashion. Must
* not be called if an error has been indicated earlier. After
* calling this, no further reading from the stream can be performed.
*
* @return False if an error occured.
*/
void Close();
/**
* Force trigger an update of the input stream.
* The action that will be taken depends on the current read mode and the
* individual input backend
* Force trigger an update of the input stream. The action that will
* be taken depends on the current read mode and the individual input
* backend.
*
* An backend can choose to ignore this.
*
@ -84,16 +81,17 @@ public:
/**
* Disables the frontend that has instantiated this backend. Once
* disabled,the frontend will not send any further message over.
* disabled, the frontend will not send any further message over.
*/
void DisableFrontend();
protected:
// Methods that have to be overwritten by the individual readers
// Methods that have to be overwritten by the individual readers
/**
* Reader-specific intialization method. Note that data may only be
* read from the input source after the Start function has been called.
* read from the input source after the Init() function has been
* called.
*
* A reader implementation must override this method. If it returns
* false, it will be assumed that a fatal error has occured that
@ -108,25 +106,25 @@ protected:
* termination.
*
* A reader implementation must override this method but it can just
* ignore calls if an input source must not be closed.
* ignore calls if an input source can't actually be closed.
*
* After the method is called, the writer will be deleted. If an error occurs
* during shutdown, an implementation should also call Error() to indicate what
* happened.
* After the method is called, the writer will be deleted. If an
* error occurs during shutdown, an implementation should also call
* Error() to indicate what happened.
*/
virtual void DoClose() = 0;
/**
* Reader-specific method implementing the forced update trigger
* Reader-specific method implementing the forced update trigger.
*
* A reader implementation must override this method but it can just ignore
* calls, if a forced update does not fit the input source or the current input
* reading mode.
* A reader implementation must override this method but it can just
* ignore calls if a forced update does not fit the input source or
* the current input reading mode.
*
* If it returns false, it will be assumed that a fatal error has occured
* that prevents the reader from further operation; it will then be
* disabled and eventually deleted. When returning false, an implementation
* should also call Error to indicate what happened.
* If it returns false, it will be assumed that a fatal error has
* occured that prevents the reader from further operation; it will
* then be disabled and eventually deleted. When returning false, an
* implementation should also call Error to indicate what happened.
*/
virtual bool DoUpdate() = 0;
@ -136,8 +134,8 @@ protected:
const string Source() const { return source; }
/**
* Method allowing a reader to send a specified bro event.
* Vals must match the values expected by the bro event.
* Method allowing a reader to send a specified Bro event. Vals must
* match the values expected by the bro event.
*
* @param name name of the bro event to send
*
@ -147,30 +145,33 @@ protected:
*/
void SendEvent(const string& name, const int num_vals, threading::Value* *vals);
// Content-sending-functions (simple mode). Including table-specific stuff that
// simply is not used if we have no table
// Content-sending-functions (simple mode). Including table-specific
// stuff that simply is not used if we have no table.
/**
* Method allowing a reader to send a list of values read for a specific stream
* back to the manager.
* Method allowing a reader to send a list of values read from a
* specific stream back to the manager in simple mode.
*
* If the stream is a table stream, the values are inserted into the table;
* if it is an event stream, the event is raised.
* If the stream is a table stream, the values are inserted into the
* table; if it is an event stream, the event is raised.
*
* @param val list of threading::Values expected by the stream
*/
void Put(threading::Value* *val);
/**
* Method allowing a reader to delete a specific value from a bro table.
* Method allowing a reader to delete a specific value from a Bro
* table.
*
* If the receiving stream is an event stream, only a removed event is raised
* If the receiving stream is an event stream, only a removed event
* is raised.
*
* @param val list of threading::Values expected by the stream
*/
void Delete(threading::Value* *val);
/**
* Method allowing a reader to clear a value from a bro table.
* Method allowing a reader to clear a Bro table.
*
* If the receiving stream is an event stream, this is ignored.
*
@ -179,25 +180,24 @@ protected:
// Content-sending-functions (tracking mode): Only changed lines are propagated.
/**
* Method allowing a reader to send a list of values read for a specific stream
* back to the manager.
* Method allowing a reader to send a list of values read from
* specific stream back to the manager in tracking mode.
*
* If the stream is a table stream, the values are inserted into the table;
* if it is an event stream, the event is raised.
* If the stream is a table stream, the values are inserted into the
* table; if it is an event stream, the event is raised.
*
* @param val list of threading::Values expected by the stream
*/
void SendEntry(threading::Value* *vals);
/**
* Method telling the manager, that the current list of entries sent by SendEntry
* is finished.
*
* For table streams, all entries that were not updated since the last EndCurrentSend
* will be deleted, because they are no longer present in the input source
* Method telling the manager, that the current list of entries sent
* by SendEntry is finished.
*
* For table streams, all entries that were not updated since the
* last EndCurrentSend will be deleted, because they are no longer
* present in the input source
*/
void EndCurrentSend();
@ -224,18 +224,14 @@ protected:
threading::Value::addr_t StringToAddr(const string &addr);
private:
// Frontend that instantiated us. This object must not be access from
// this class, it's running in a different thread!
// Frontend that instantiated us. This object must not be accessed
// from this class, it's running in a different thread!
ReaderFrontend* frontend;
string source;
bool disabled;
// For implementing Fmt().
char* buf;
unsigned int buf_len;
unsigned int num_fields;
const threading::Field* const * fields; // raw mapping
};

12
src/input/ReaderFrontend.cc
View file

@ -12,11 +12,15 @@ namespace input {
class InitMessage : public threading::InputMessage<ReaderBackend>
{
public:
InitMessage(ReaderBackend* backend, const string source, const int mode, const int num_fields, const threading::Field* const* fields)
InitMessage(ReaderBackend* backend, const string source, const int mode,
const int num_fields, const threading::Field* const* fields)
: threading::InputMessage<ReaderBackend>("Init", backend),
source(source), mode(mode), num_fields(num_fields), fields(fields) { }
virtual bool Process() { return Object()->Init(source, mode, num_fields, fields); }
virtual bool Process()
{
return Object()->Init(source, mode, num_fields, fields);
}
private:
const string source;
@ -80,7 +84,7 @@ void ReaderFrontend::Update()
if ( disabled )
return;
if ( !initialized )
if ( ! initialized )
{
reporter->Error("Tried to call update on uninitialized reader");
return;
@ -94,7 +98,7 @@ void ReaderFrontend::Close()
if ( disabled )
return;
if ( !initialized )
if ( ! initialized )
{
reporter->Error("Tried to call finish on uninitialized reader");
return;

30
src/input/ReaderFrontend.h
View file

@ -47,37 +47,39 @@ public:
* sends a message back that will asynchronously call Disable().
*
* See ReaderBackend::Init() for arguments.
*
* This method must only be called from the main thread.
*/
void Init(string arg_source, int mode, const int arg_num_fields, const threading::Field* const* fields);
/**
* Force an update of the current input source. Actual action depends on
* the opening mode and on the input source.
* Force an update of the current input source. Actual action depends
* on the opening mode and on the input source.
*
* This method generates a message to the backend reader and triggers
* the corresponding message there.
*
* This method must only be called from the main thread.
*/
void Update();
/**
* Finalizes writing to this tream.
* Finalizes reading from this stream.
*
* This method generates a message to the backend reader and triggers
* the corresponding message there.
* This method must only be called from the main thread.
* the corresponding message there. This method must only be called
* from the main thread.
*/
void Close();
/**
* Disables the reader frontend. From now on, all method calls that
* would normally send message over to the backend, turn into no-ops.
* Note though that it does not stop the backend itself, use Finsh()
* Note though that it does not stop the backend itself, use Finish()
* to do that as well (this method is primarily for use as callback
* when the backend wants to disable the frontend).
*
* Disabled frontend will eventually be discarded by the
* Disabled frontends will eventually be discarded by the
* input::Manager.
*
* This method must only be called from the main thread.
@ -85,7 +87,8 @@ public:
void SetDisable() { disabled = true; }
/**
* Returns true if the reader frontend has been disabled with SetDisable().
* Returns true if the reader frontend has been disabled with
* SetDisable().
*/
bool Disabled() { return disabled; }
@ -101,18 +104,21 @@ protected:
friend class Manager;
/**
* Returns the source as passed into the constructor
* Returns the source as passed into the constructor.
*/
const string Source() const { return source; };
const string& Source() const { return source; };
string ty_name; // Name of the backend type. Set by the manager.
/**
* Returns the name of the backend's type.
*/
const string& TypeName() const { return ty_name; }
private:
ReaderBackend* backend; // The backend we have instanatiated.
string source;
string ty_name; // Backend type, set by manager.
bool disabled; // True if disabled.
bool initialized; // True if initialized.
};
}

63
src/input/readers/Ascii.cc
View file

@ -20,7 +20,6 @@ using namespace input::reader;
using threading::Value;
using threading::Field;
FieldMapping::FieldMapping(const string& arg_name, const TypeTag& arg_type, int arg_position)
: name(arg_name), type(arg_type)
{
@ -54,14 +53,15 @@ Ascii::Ascii(ReaderFrontend *frontend) : ReaderBackend(frontend)
{
file = 0;
separator.assign( (const char*) BifConst::InputAscii::separator->Bytes(),
BifConst::InputAscii::separator->Len());
if ( separator.size() != 1 )
Error("separator length has to be 1. Separator will be truncated.");
set_separator.assign( (const char*) BifConst::InputAscii::set_separator->Bytes(),
BifConst::InputAscii::set_separator->Len());
if ( set_separator.size() != 1 )
Error("set_separator length has to be 1. Separator will be truncated.");
@ -70,7 +70,6 @@ Ascii::Ascii(ReaderFrontend *frontend) : ReaderBackend(frontend)
unset_field.assign( (const char*) BifConst::InputAscii::unset_field->Bytes(),
BifConst::InputAscii::unset_field->Len());
}
Ascii::~Ascii()
@ -97,14 +96,14 @@ bool Ascii::DoInit(string path, int arg_mode, int arg_num_fields, const Field* c
num_fields = arg_num_fields;
fields = arg_fields;
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) )
if ( (mode != MANUAL) && (mode != REREAD) && (mode != STREAM) )
{
Error(Fmt("Unsupported read mode %d for source %s", mode, path.c_str()));
return false;
}
file = new ifstream(path.c_str());
if ( !file->is_open() )
if ( ! file->is_open() )
{
Error(Fmt("Init: cannot open %s", fname.c_str()));
delete(file);
@ -133,9 +132,9 @@ bool Ascii::ReadHeader(bool useCached)
string line;
map<string, uint32_t> ifields;
if ( !useCached )
if ( ! useCached )
{
if ( !GetLine(line) )
if ( ! GetLine(line) )
{
Error("could not read first line");
return false;
@ -143,6 +142,7 @@ bool Ascii::ReadHeader(bool useCached)
headerline = line;
}
else
line = headerline;
@ -152,7 +152,7 @@ bool Ascii::ReadHeader(bool useCached)
while ( splitstream )
{
string s;
if ( !getline(splitstream, s, separator[0]))
if ( ! getline(splitstream, s, separator[0]))
break;
ifields[s] = pos;
@ -178,22 +178,27 @@ bool Ascii::ReadHeader(bool useCached)
continue;
}
Error(Fmt("Did not find requested field %s in input data file %s.", field->name.c_str(), fname.c_str()));
Error(Fmt("Did not find requested field %s in input data file %s.",
field->name.c_str(), fname.c_str()));
return false;
}
FieldMapping f(field->name, field->type, field->subtype, ifields[field->name]);
if ( field->secondary_name != "" )
{
map<string, uint32_t>::iterator fit2 = ifields.find(field->secondary_name);
if ( fit2 == ifields.end() )
{
Error(Fmt("Could not find requested port type field %s in input data file.", field->secondary_name.c_str()));
Error(Fmt("Could not find requested port type field %s in input data file.",
field->secondary_name.c_str()));
return false;
}
f.secondary_position = ifields[field->secondary_name];
}
columnMap.push_back(f);
}
@ -222,7 +227,6 @@ bool Ascii::GetLine(string& str)
Value* Ascii::EntryToVal(string s, FieldMapping field)
{
if ( s.compare(unset_field) == 0 ) // field is not set...
return new Value(field.type, false);
@ -241,7 +245,8 @@ Value* Ascii::EntryToVal(string s, FieldMapping field)
val->val.int_val = 0;
else
{
Error(Fmt("Field: %s Invalid value for boolean: %s", field.name.c_str(), s.c_str()));
Error(Fmt("Field: %s Invalid value for boolean: %s",
field.name.c_str(), s.c_str()));
return false;
}
break;
@ -269,10 +274,12 @@ Value* Ascii::EntryToVal(string s, FieldMapping field)
case TYPE_SUBNET:
{
size_t pos = s.find("/");
if ( pos == s.npos ) {
if ( pos == s.npos )
{
Error(Fmt("Invalid value for subnet: %s", s.c_str()));
return false;
}
}
int width = atoi(s.substr(pos+1).c_str());
string addr = s.substr(0, pos);
@ -295,7 +302,7 @@ Value* Ascii::EntryToVal(string s, FieldMapping field)
// how many entries do we have...
unsigned int length = 1;
for ( unsigned int i = 0; i < s.size(); i++ )
if ( s[i] == ',') length++;
if ( s[i] == ',' ) length++;
unsigned int pos = 0;
@ -309,15 +316,15 @@ Value* Ascii::EntryToVal(string s, FieldMapping field)
val->val.set_val.vals = lvals;
val->val.set_val.size = length;
}
else if ( field.type == TYPE_VECTOR )
{
val->val.vector_val.vals = lvals;
val->val.vector_val.size = length;
}
else
{
assert(false);
}
if ( length == 0 )
break; //empty
@ -327,7 +334,7 @@ Value* Ascii::EntryToVal(string s, FieldMapping field)
{
string element;
if ( !getline(splitstream, element, set_separator[0]) )
if ( ! getline(splitstream, element, set_separator[0]) )
break;
if ( pos >= length )
@ -343,12 +350,12 @@ Value* Ascii::EntryToVal(string s, FieldMapping field)
Error("Error while reading set");
return 0;
}
lvals[pos] = newval;
pos++;
}
if ( pos != length )
{
Error("Internal error while parsing set: did not find all elements");
@ -358,7 +365,6 @@ Value* Ascii::EntryToVal(string s, FieldMapping field)
break;
}
default:
Error(Fmt("unsupported field format %d for %s", field.type,
field.name.c_str()));
@ -373,6 +379,7 @@ bool Ascii::DoUpdate()
{
switch ( mode ) {
case REREAD:
{
// check if the file has changed
struct stat sb;
if ( stat(fname.c_str(), &sb) == -1 )
@ -388,9 +395,11 @@ bool Ascii::DoUpdate()
// file changed. reread.
// fallthrough
}
case MANUAL:
case STREAM:
{
// dirty, fix me. (well, apparently after trying seeking, etc
// - this is not that bad)
if ( file && file->is_open() )
@ -405,6 +414,7 @@ bool Ascii::DoUpdate()
}
file->close();
}
file = new ifstream(fname.c_str());
if ( !file->is_open() )
{
@ -412,13 +422,14 @@ bool Ascii::DoUpdate()
return false;
}
if ( ReadHeader(false) == false )
{
return false;
}
break;
}
default:
assert(false);
@ -435,7 +446,7 @@ bool Ascii::DoUpdate()
while ( splitstream )
{
string s;
if ( !getline(splitstream, s, separator[0]) )
if ( ! getline(splitstream, s, separator[0]) )
break;
stringfields[pos] = s;
@ -444,7 +455,6 @@ bool Ascii::DoUpdate()
pos--; // for easy comparisons of max element.
Value** fields = new Value*[num_fields];
int fpos = 0;
@ -465,7 +475,8 @@ bool Ascii::DoUpdate()
if ( (*fit).position > pos || (*fit).secondary_position > pos )
{
Error(Fmt("Not enough fields in line %s. Found %d fields, want positions %d and %d", line.c_str(), pos, (*fit).position, (*fit).secondary_position));
Error(Fmt("Not enough fields in line %s. Found %d fields, want positions %d and %d",
line.c_str(), pos, (*fit).position, (*fit).secondary_position));
return false;
}
@ -513,11 +524,13 @@ bool Ascii::DoHeartbeat(double network_time, double current_time)
case MANUAL:
// yay, we do nothing :)
break;
case REREAD:
case STREAM:
Update(); // call update and not DoUpdate, because update
// checks disabled.
break;
default:
assert(false);
}

56
src/input/readers/Ascii.h
View file

@ -14,11 +14,9 @@ namespace input { namespace reader {
struct FieldMapping {
string name;
TypeTag type;
// internal type for sets and vectors
TypeTag subtype;
TypeTag subtype; // internal type for sets and vectors
int position;
// for ports: pos of the second field
int secondary_position;
int secondary_position; // for ports: pos of the second field
bool present;
FieldMapping(const string& arg_name, const TypeTag& arg_type, int arg_position);
@ -27,60 +25,46 @@ struct FieldMapping {
FieldMapping() { position = -1; secondary_position = -1; }
FieldMapping subType();
//bool IsEmpty() { return position == -1; }
};
class Ascii : public ReaderBackend {
public:
Ascii(ReaderFrontend* frontend);
~Ascii();
Ascii(ReaderFrontend* frontend);
~Ascii();
static ReaderBackend* Instantiate(ReaderFrontend* frontend) { return new Ascii(frontend); }
static ReaderBackend* Instantiate(ReaderFrontend* frontend) { return new Ascii(frontend); }
protected:
virtual bool DoInit(string path, int mode, int arg_num_fields, const threading::Field* const* fields);
virtual void DoClose();
virtual bool DoUpdate();
private:
virtual bool DoHeartbeat(double network_time, double current_time);
unsigned int num_fields;
bool ReadHeader(bool useCached);
bool GetLine(string& str);
threading::Value* EntryToVal(string s, FieldMapping type);
const threading::Field* const * fields; // raw mapping
unsigned int num_fields;
const threading::Field* const *fields; // raw mapping
ifstream* file;
string fname;
int mode;
time_t mtime;
// map columns in the file to columns to send back to the manager
vector<FieldMapping> columnMap;
bool ReadHeader(bool useCached);
threading::Value* EntryToVal(string s, FieldMapping type);
bool GetLine(string& str);
ifstream* file;
string fname;
// Options set from the script-level.
string separator;
string set_separator;
string empty_field;
string unset_field;
// keep a copy of the headerline to determine field locations when stream descriptions change
string headerline;
int mode;
time_t mtime;
// Options set from the script-level.
string separator;
string set_separator;
string empty_field;
string unset_field;
};

10
src/input/readers/Benchmark.cc
View file

@ -31,7 +31,6 @@ Benchmark::Benchmark(ReaderFrontend *frontend) : ReaderBackend(frontend)
stopspreadat = int(BifConst::InputBenchmark::stopspreadat);
timedspread = double(BifConst::InputBenchmark::timedspread);
heart_beat_interval = double(BifConst::Threading::heart_beat_interval);
}
Benchmark::~Benchmark()
@ -54,7 +53,7 @@ bool Benchmark::DoInit(string path, int arg_mode, int arg_num_fields, const Fiel
if ( autospread != 0.0 )
autospread_time = (int) ( (double) 1000000 / (autospread * (double) num_lines) );
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) )
if ( (mode != MANUAL) && (mode != REREAD) && (mode != STREAM) )
{
Error(Fmt("Unsupported read mode %d for source %s", mode, path.c_str()));
return false;
@ -76,7 +75,7 @@ string Benchmark::RandomString(const int len)
"abcdefghijklmnopqrstuvwxyz";
for (int i = 0; i < len; ++i)
s[i] = values[rand() / (RAND_MAX / sizeof(values))];
s[i] = values[rand() / (RAND_MAX / sizeof(values))];
return s;
}
@ -93,7 +92,7 @@ double Benchmark::CurrTime()
// read the entire file and send appropriate thingies back to InputMgr
bool Benchmark::DoUpdate()
{
int linestosend = num_lines * heart_beat_interval;
int linestosend = num_lines * heart_beat_interval;
for ( int i = 0; i < linestosend; i++ )
{
Value** field = new Value*[num_fields];
@ -141,6 +140,7 @@ threading::Value* Benchmark::EntryToVal(TypeTag type, TypeTag subtype)
switch ( type ) {
case TYPE_ENUM:
assert(false); // no enums, please.
case TYPE_STRING:
val->val.string_val = new string(RandomString(10));
break;
@ -247,6 +247,7 @@ bool Benchmark::DoHeartbeat(double network_time, double current_time)
case MANUAL:
// yay, we do nothing :)
break;
case REREAD:
case STREAM:
if ( multiplication_factor != 1 || add != 0 )
@ -275,4 +276,3 @@ bool Benchmark::DoHeartbeat(double network_time, double current_time)
return true;
}

27
src/input/readers/Benchmark.h
View file

@ -3,41 +3,37 @@
#ifndef INPUT_READERS_BENCHMARK_H
#define INPUT_READERS_BENCHMARK_H
#include "../ReaderBackend.h"
namespace input { namespace reader {
/**
* A benchmark reader to measure performance of the input framework.
*/
class Benchmark : public ReaderBackend {
public:
Benchmark(ReaderFrontend* frontend);
~Benchmark();
Benchmark(ReaderFrontend* frontend);
~Benchmark();
static ReaderBackend* Instantiate(ReaderFrontend* frontend) { return new Benchmark(frontend); }
static ReaderBackend* Instantiate(ReaderFrontend* frontend) { return new Benchmark(frontend); }
protected:
virtual bool DoInit(string path, int mode, int arg_num_fields, const threading::Field* const* fields);
virtual void DoClose();
virtual bool DoUpdate();
private:
virtual bool DoHeartbeat(double network_time, double current_time);
unsigned int num_fields;
double CurrTime();
const threading::Field* const * fields; // raw mapping
string RandomString(const int len);
threading::Value* EntryToVal(TypeTag Type, TypeTag subtype);
unsigned int num_fields;
const threading::Field* const * fields; // raw mapping
int mode;
int num_lines;
double multiplication_factor;
int spread;
double autospread;
@ -47,9 +43,6 @@ private:
double heartbeatstarttime;
double timedspread;
double heart_beat_interval;
string RandomString(const int len);
};

91
src/input/readers/Raw.cc
View file

@ -28,7 +28,9 @@ Raw::Raw(ReaderFrontend *frontend) : ReaderBackend(frontend)
file = 0;
in = 0;
separator.assign( (const char*) BifConst::InputRaw::record_separator->Bytes(), BifConst::InputRaw::record_separator->Len());
separator.assign( (const char*) BifConst::InputRaw::record_separator->Bytes(),
BifConst::InputRaw::record_separator->Len());
if ( separator.size() != 1 )
Error("separator length has to be 1. Separator will be truncated.");
}
@ -67,12 +69,11 @@ bool Raw::Open()
}
}
// This is defined in input/fdstream.h
in = new boost::fdistream(fileno(file));
if ( execute && mode == STREAM )
{
fcntl(fileno(file), F_SETFL, O_NONBLOCK);
}
return true;
}
@ -141,8 +142,9 @@ bool Raw::DoInit(string path, int arg_mode, int arg_num_fields, const Field* con
execute = true;
fname = path.substr(0, fname.length() - 1);
if ( ( mode != MANUAL ) && ( mode != STREAM ) ) {
Error(Fmt("Unsupported read mode %d for source %s in execution mode", mode, fname.c_str()));
if ( (mode != MANUAL) && (mode != STREAM) ) {
Error(Fmt("Unsupported read mode %d for source %s in execution mode",
mode, fname.c_str()));
return false;
}
@ -150,9 +152,10 @@ bool Raw::DoInit(string path, int arg_mode, int arg_num_fields, const Field* con
} else {
execute = false;
if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) )
if ( (mode != MANUAL) && (mode != REREAD) && (mode != STREAM) )
{
Error(Fmt("Unsupported read mode %d for source %s", mode, fname.c_str()));
Error(Fmt("Unsupported read mode %d for source %s",
mode, fname.c_str()));
return false;
}
@ -184,57 +187,55 @@ bool Raw::GetLine(string& str)
if ( in->eofbit == true || in->failbit == true )
return false;
while ( getline(*in, str, separator[0]) )
return true;
return false;
return getline(*in, str, separator[0]);
}
// read the entire file and send appropriate thingies back to InputMgr
bool Raw::DoUpdate()
{
if ( firstrun )
firstrun = false;
else
{
switch ( mode ) {
case REREAD:
case REREAD:
{
// check if the file has changed
struct stat sb;
if ( stat(fname.c_str(), &sb) == -1 )
{
// check if the file has changed
struct stat sb;
if ( stat(fname.c_str(), &sb) == -1 )
{
Error(Fmt("Could not get stat for %s", fname.c_str()));
return false;
}
if ( sb.st_mtime <= mtime )
// no change
return true;
mtime = sb.st_mtime;
// file changed. reread.
// fallthrough
Error(Fmt("Could not get stat for %s", fname.c_str()));
return false;
}
case MANUAL:
case STREAM:
if ( mode == STREAM && file != NULL && in != NULL )
{
//fpurge(file);
in->clear(); // remove end of file evil bits
break;
}
Close();
if ( !Open() )
return false;
if ( sb.st_mtime <= mtime )
// no change
return true;
mtime = sb.st_mtime;
// file changed. reread.
//
// fallthrough
}
case MANUAL:
case STREAM:
if ( mode == STREAM && file != NULL && in != NULL )
{
//fpurge(file);
in->clear(); // remove end of file evil bits
break;
default:
assert(false);
}
Close();
if ( ! Open() )
return false;
break;
default:
assert(false);
}
}
@ -260,7 +261,6 @@ bool Raw::DoUpdate()
return true;
}
bool Raw::DoHeartbeat(double network_time, double current_time)
{
ReaderBackend::DoHeartbeat(network_time, current_time);
@ -269,10 +269,11 @@ bool Raw::DoHeartbeat(double network_time, double current_time)
case MANUAL:
// yay, we do nothing :)
break;
case REREAD:
case STREAM:
Update(); // call update and not DoUpdate, because update
// checks disabled.
Update(); // call update and not DoUpdate, because update
// checks disabled.
break;
default:
assert(false);

33
src/input/readers/Raw.h
View file

@ -10,51 +10,44 @@
namespace input { namespace reader {
/**
* A reader that returns a file (or the output of a command) as a single
* blob.
*/
class Raw : public ReaderBackend {
public:
Raw(ReaderFrontend* frontend);
~Raw();
Raw(ReaderFrontend* frontend);
~Raw();
static ReaderBackend* Instantiate(ReaderFrontend* frontend) { return new Raw(frontend); }
static ReaderBackend* Instantiate(ReaderFrontend* frontend) { return new Raw(frontend); }
protected:
virtual bool DoInit(string path, int mode, int arg_num_fields, const threading::Field* const* fields);
virtual void DoClose();
virtual bool DoUpdate();
private:
virtual bool DoHeartbeat(double network_time, double current_time);
bool Open();
bool Close();
bool GetLine(string& str);
unsigned int num_fields;
const threading::Field* const * fields; // raw mapping
istream* in;
FILE* file;
string fname;
// Options set from the script-level.
string separator;
int mode;
bool execute;
bool firstrun;
time_t mtime;
unsigned int num_fields;
const threading::Field* const * fields; // raw mapping
// Options set from the script-level.
string separator;
};
}
}

2
testing/btest/Baseline/scripts.base.frameworks.input.executeraw/out
View file

@ -6,4 +6,4 @@ print outfile, s;
close(outfile);
}]
Input::EVENT_NEW
8 ../input.log
8 ../input.log

3
testing/btest/scripts/base/frameworks/input/executeraw.bro
View file

@ -1,6 +1,7 @@
#
# @TEST-EXEC: btest-bg-run bro bro -b %INPUT
# @TEST-EXEC: btest-bg-wait -k 1
# @TEST-EXEC: cat out.tmp | sed 's/^ *//g' >out
# @TEST-EXEC: btest-diff out
@TEST-START-FILE input.log
@ -31,7 +32,7 @@ event line(description: Input::EventDescription, tpe: Input::Event, s: string) {
event bro_init()
{
outfile = open ("../out");
outfile = open ("../out.tmp");
Input::add_event([$source="wc -l ../input.log |", $reader=Input::READER_RAW, $name="input", $fields=Val, $ev=line]);
Input::remove("input");
}