From 26f78d7f1b12a0590b51ea8b3a65b45784783907 Mon Sep 17 00:00:00 2001 From: Goutham Veeramachaneni <gouthamve@gmail.com> Date: Tue, 4 Dec 2018 15:49:57 +0530 Subject: [PATCH] Support chained filtering Signed-off-by: Goutham Veeramachaneni <gouthamve@gmail.com> --- pkg/ingester/instance.go | 21 +- pkg/iter/iterator.go | 66 +++++- pkg/logproto/logproto.pb.go | 391 ++++++++++++++++++++++++++++++------ pkg/logproto/logproto.proto | 8 +- pkg/querier/http.go | 21 +- 5 files changed, 433 insertions(+), 74 deletions(-) diff --git a/pkg/ingester/instance.go b/pkg/ingester/instance.go index ff7af878..8ae9d495 100644 --- a/pkg/ingester/instance.go +++ b/pkg/ingester/instance.go @@ -109,12 +109,23 @@ func (i *instance) Query(req *logproto.QueryRequest, queryServer logproto.Querie iterator := iter.NewHeapIterator(iterators, req.Direction) defer helpers.LogError("closing iterator", iterator.Close) - if req.Regex != "" { - var err error - iterator, err = iter.NewRegexpFilter(req.Regex, iterator) - if err != nil { - return err + if len(req.Filters) > 0 { + filters := make([]iter.Filter, 0, len(req.Filters)) + for _, filter := range req.Filters { + f, err := iter.NewRegexpFilter(filter.Pattern) + if err != nil { + return err + } + + if filter.Inverse { + f = iter.NewInverseFilter(f) + } + + filters = append(filters, f) } + + filter := iter.NewChainedFilter(filters...) + iterator = iter.NewFilteredIterator(filter, iterator) } return sendBatches(iterator, queryServer, req.Limit) diff --git a/pkg/iter/iterator.go b/pkg/iter/iterator.go index e5862290..189fd5f6 100644 --- a/pkg/iter/iterator.go +++ b/pkg/iter/iterator.go @@ -243,29 +243,83 @@ func (i *queryClientIterator) Close() error { return i.client.CloseSend() } +// Filter is used to filter the log lines we want. +type Filter interface { + Filter(string) bool +} + type regexpFilter struct { re *regexp.Regexp - EntryIterator } // NewRegexpFilter returns an iterator that filters entries by regexp. -func NewRegexpFilter(r string, i EntryIterator) (EntryIterator, error) { +func NewRegexpFilter(r string) (Filter, error) { re, err := regexp.Compile(r) if err != nil { return nil, err } return ®expFilter{ - re: re, - EntryIterator: i, + re: re, }, nil } -func (i *regexpFilter) Next() bool { +func (f *regexpFilter) Filter(log string) bool { + return f.re.MatchString(log) +} + +type inverseFilter struct { + filter Filter +} + +// NewInverseFilter filter returns an inverse filter. +func NewInverseFilter(f Filter) Filter { + return inverseFilter{filter: f} +} + +func (f inverseFilter) Filter(log string) bool { + return !f.filter.Filter(log) +} + +type chainedFilter struct { + filters []Filter +} + +// NewChainedFilter returns a filter that chains the filters with AND. +func NewChainedFilter(filters ...Filter) Filter { + return chainedFilter{filters: filters} +} + +func (f chainedFilter) Filter(log string) bool { + for _, filter := range f.filters { + if !filter.Filter(log) { + return false + } + } + + return true +} + +type filteredIterator struct { + EntryIterator + + filter Filter +} + +// NewFilteredIterator returns an iterator with the filter applied. +func NewFilteredIterator(filter Filter, it EntryIterator) EntryIterator { + return &filteredIterator{ + EntryIterator: it, + filter: filter, + } +} + +func (i *filteredIterator) Next() bool { for i.EntryIterator.Next() { - if i.re.MatchString(i.Entry().Line) { + if i.filter.Filter(i.Entry().Line) { return true } } + return false } diff --git a/pkg/logproto/logproto.pb.go b/pkg/logproto/logproto.pb.go index 155b48f9..06f99299 100644 --- a/pkg/logproto/logproto.pb.go +++ b/pkg/logproto/logproto.pb.go @@ -54,7 +54,7 @@ var Direction_value = map[string]int32{ } func (Direction) EnumDescriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{0} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{0} } type PushRequest struct { @@ -64,7 +64,7 @@ type PushRequest struct { func (m *PushRequest) Reset() { *m = PushRequest{} } func (*PushRequest) ProtoMessage() {} func (*PushRequest) Descriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{0} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{0} } func (m *PushRequest) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -106,7 +106,7 @@ type PushResponse struct { func (m *PushResponse) Reset() { *m = PushResponse{} } func (*PushResponse) ProtoMessage() {} func (*PushResponse) Descriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{1} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{1} } func (m *PushResponse) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -141,13 +141,13 @@ type QueryRequest struct { Start time.Time `protobuf:"bytes,3,opt,name=start,stdtime" json:"start"` End time.Time `protobuf:"bytes,4,opt,name=end,stdtime" json:"end"` Direction Direction `protobuf:"varint,5,opt,name=direction,proto3,enum=logproto.Direction" json:"direction,omitempty"` - Regex string `protobuf:"bytes,6,opt,name=regex,proto3" json:"regex,omitempty"` + Filters []*Filter `protobuf:"bytes,6,rep,name=filters" json:"filters,omitempty"` } func (m *QueryRequest) Reset() { *m = QueryRequest{} } func (*QueryRequest) ProtoMessage() {} func (*QueryRequest) Descriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{2} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{2} } func (m *QueryRequest) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -211,13 +211,64 @@ func (m *QueryRequest) GetDirection() Direction { return FORWARD } -func (m *QueryRequest) GetRegex() string { +func (m *QueryRequest) GetFilters() []*Filter { if m != nil { - return m.Regex + return m.Filters + } + return nil +} + +type Filter struct { + Pattern string `protobuf:"bytes,1,opt,name=pattern,proto3" json:"pattern,omitempty"` + Inverse bool `protobuf:"varint,2,opt,name=inverse,proto3" json:"inverse,omitempty"` +} + +func (m *Filter) Reset() { *m = Filter{} } +func (*Filter) ProtoMessage() {} +func (*Filter) Descriptor() ([]byte, []int) { + return fileDescriptor_logproto_5d6f332523c5fa62, []int{3} +} +func (m *Filter) XXX_Unmarshal(b []byte) error { + return m.Unmarshal(b) +} +func (m *Filter) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) { + if deterministic { + return xxx_messageInfo_Filter.Marshal(b, m, deterministic) + } else { + b = b[:cap(b)] + n, err := m.MarshalTo(b) + if err != nil { + return nil, err + } + return b[:n], nil + } +} +func (dst *Filter) XXX_Merge(src proto.Message) { + xxx_messageInfo_Filter.Merge(dst, src) +} +func (m *Filter) XXX_Size() int { + return m.Size() +} +func (m *Filter) XXX_DiscardUnknown() { + xxx_messageInfo_Filter.DiscardUnknown(m) +} + +var xxx_messageInfo_Filter proto.InternalMessageInfo + +func (m *Filter) GetPattern() string { + if m != nil { + return m.Pattern } return "" } +func (m *Filter) GetInverse() bool { + if m != nil { + return m.Inverse + } + return false +} + type QueryResponse struct { Streams []*Stream `protobuf:"bytes,1,rep,name=streams" json:"streams,omitempty"` } @@ -225,7 +276,7 @@ type QueryResponse struct { func (m *QueryResponse) Reset() { *m = QueryResponse{} } func (*QueryResponse) ProtoMessage() {} func (*QueryResponse) Descriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{3} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{4} } func (m *QueryResponse) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -269,7 +320,7 @@ type LabelRequest struct { func (m *LabelRequest) Reset() { *m = LabelRequest{} } func (*LabelRequest) ProtoMessage() {} func (*LabelRequest) Descriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{4} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{5} } func (m *LabelRequest) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -319,7 +370,7 @@ type LabelResponse struct { func (m *LabelResponse) Reset() { *m = LabelResponse{} } func (*LabelResponse) ProtoMessage() {} func (*LabelResponse) Descriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{5} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{6} } func (m *LabelResponse) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -363,7 +414,7 @@ type Stream struct { func (m *Stream) Reset() { *m = Stream{} } func (*Stream) ProtoMessage() {} func (*Stream) Descriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{6} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{7} } func (m *Stream) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -414,7 +465,7 @@ type Entry struct { func (m *Entry) Reset() { *m = Entry{} } func (*Entry) ProtoMessage() {} func (*Entry) Descriptor() ([]byte, []int) { - return fileDescriptor_logproto_09f5536631d1d06a, []int{7} + return fileDescriptor_logproto_5d6f332523c5fa62, []int{8} } func (m *Entry) XXX_Unmarshal(b []byte) error { return m.Unmarshal(b) @@ -461,6 +512,7 @@ func init() { proto.RegisterType((*PushRequest)(nil), "logproto.PushRequest") proto.RegisterType((*PushResponse)(nil), "logproto.PushResponse") proto.RegisterType((*QueryRequest)(nil), "logproto.QueryRequest") + proto.RegisterType((*Filter)(nil), "logproto.Filter") proto.RegisterType((*QueryResponse)(nil), "logproto.QueryResponse") proto.RegisterType((*LabelRequest)(nil), "logproto.LabelRequest") proto.RegisterType((*LabelResponse)(nil), "logproto.LabelResponse") @@ -559,7 +611,39 @@ func (this *QueryRequest) Equal(that interface{}) bool { if this.Direction != that1.Direction { return false } - if this.Regex != that1.Regex { + if len(this.Filters) != len(that1.Filters) { + return false + } + for i := range this.Filters { + if !this.Filters[i].Equal(that1.Filters[i]) { + return false + } + } + return true +} +func (this *Filter) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + that1, ok := that.(*Filter) + if !ok { + that2, ok := that.(Filter) + if ok { + that1 = &that2 + } else { + return false + } + } + if that1 == nil { + return this == nil + } else if this == nil { + return false + } + if this.Pattern != that1.Pattern { + return false + } + if this.Inverse != that1.Inverse { return false } return true @@ -740,7 +824,20 @@ func (this *QueryRequest) GoString() string { s = append(s, "Start: "+fmt.Sprintf("%#v", this.Start)+",\n") s = append(s, "End: "+fmt.Sprintf("%#v", this.End)+",\n") s = append(s, "Direction: "+fmt.Sprintf("%#v", this.Direction)+",\n") - s = append(s, "Regex: "+fmt.Sprintf("%#v", this.Regex)+",\n") + if this.Filters != nil { + s = append(s, "Filters: "+fmt.Sprintf("%#v", this.Filters)+",\n") + } + s = append(s, "}") + return strings.Join(s, "") +} +func (this *Filter) GoString() string { + if this == nil { + return "nil" + } + s := make([]string, 0, 6) + s = append(s, "&logproto.Filter{") + s = append(s, "Pattern: "+fmt.Sprintf("%#v", this.Pattern)+",\n") + s = append(s, "Inverse: "+fmt.Sprintf("%#v", this.Inverse)+",\n") s = append(s, "}") return strings.Join(s, "") } @@ -1106,11 +1203,51 @@ func (m *QueryRequest) MarshalTo(dAtA []byte) (int, error) { i++ i = encodeVarintLogproto(dAtA, i, uint64(m.Direction)) } - if len(m.Regex) > 0 { - dAtA[i] = 0x32 + if len(m.Filters) > 0 { + for _, msg := range m.Filters { + dAtA[i] = 0x32 + i++ + i = encodeVarintLogproto(dAtA, i, uint64(msg.Size())) + n, err := msg.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n + } + } + return i, nil +} + +func (m *Filter) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *Filter) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Pattern) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintLogproto(dAtA, i, uint64(len(m.Pattern))) + i += copy(dAtA[i:], m.Pattern) + } + if m.Inverse { + dAtA[i] = 0x10 + i++ + if m.Inverse { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } i++ - i = encodeVarintLogproto(dAtA, i, uint64(len(m.Regex))) - i += copy(dAtA[i:], m.Regex) } return i, nil } @@ -1333,10 +1470,28 @@ func (m *QueryRequest) Size() (n int) { if m.Direction != 0 { n += 1 + sovLogproto(uint64(m.Direction)) } - l = len(m.Regex) + if len(m.Filters) > 0 { + for _, e := range m.Filters { + l = e.Size() + n += 1 + l + sovLogproto(uint64(l)) + } + } + return n +} + +func (m *Filter) Size() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + l = len(m.Pattern) if l > 0 { n += 1 + l + sovLogproto(uint64(l)) } + if m.Inverse { + n += 2 + } return n } @@ -1462,7 +1617,18 @@ func (this *QueryRequest) String() string { `Start:` + strings.Replace(strings.Replace(this.Start.String(), "Timestamp", "types.Timestamp", 1), `&`, ``, 1) + `,`, `End:` + strings.Replace(strings.Replace(this.End.String(), "Timestamp", "types.Timestamp", 1), `&`, ``, 1) + `,`, `Direction:` + fmt.Sprintf("%v", this.Direction) + `,`, - `Regex:` + fmt.Sprintf("%v", this.Regex) + `,`, + `Filters:` + strings.Replace(fmt.Sprintf("%v", this.Filters), "Filter", "Filter", 1) + `,`, + `}`, + }, "") + return s +} +func (this *Filter) String() string { + if this == nil { + return "nil" + } + s := strings.Join([]string{`&Filter{`, + `Pattern:` + fmt.Sprintf("%v", this.Pattern) + `,`, + `Inverse:` + fmt.Sprintf("%v", this.Inverse) + `,`, `}`, }, "") return s @@ -1817,7 +1983,88 @@ func (m *QueryRequest) Unmarshal(dAtA []byte) error { } case 6: if wireType != 2 { - return fmt.Errorf("proto: wrong wireType = %d for field Regex", wireType) + return fmt.Errorf("proto: wrong wireType = %d for field Filters", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowLogproto + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthLogproto + } + postIndex := iNdEx + msglen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Filters = append(m.Filters, &Filter{}) + if err := m.Filters[len(m.Filters)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipLogproto(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthLogproto + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *Filter) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowLogproto + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: Filter: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: Filter: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Pattern", wireType) } var stringLen uint64 for shift := uint(0); ; shift += 7 { @@ -1842,8 +2089,28 @@ func (m *QueryRequest) Unmarshal(dAtA []byte) error { if postIndex > l { return io.ErrUnexpectedEOF } - m.Regex = string(dAtA[iNdEx:postIndex]) + m.Pattern = string(dAtA[iNdEx:postIndex]) iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Inverse", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowLogproto + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + m.Inverse = bool(v != 0) default: iNdEx = preIndex skippy, err := skipLogproto(dAtA[iNdEx:]) @@ -2448,43 +2715,45 @@ var ( ErrIntOverflowLogproto = fmt.Errorf("proto: integer overflow") ) -func init() { proto.RegisterFile("logproto.proto", fileDescriptor_logproto_09f5536631d1d06a) } - -var fileDescriptor_logproto_09f5536631d1d06a = []byte{ - // 560 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x52, 0x3b, 0x6f, 0xd4, 0x4c, - 0x14, 0xf5, 0x7c, 0xfb, 0xf4, 0xdd, 0xc7, 0x17, 0x0d, 0x10, 0xac, 0x2d, 0x66, 0x57, 0x2e, 0x60, - 0x15, 0x09, 0x2f, 0x2c, 0x08, 0x44, 0xa0, 0x59, 0x13, 0x68, 0x40, 0x82, 0x0c, 0x48, 0x94, 0xc8, - 0x9b, 0x0c, 0x8e, 0x25, 0x3f, 0x36, 0xf6, 0x18, 0x91, 0x0e, 0x89, 0x3f, 0x90, 0x9f, 0xc1, 0x4f, - 0x49, 0xb9, 0x65, 0x2a, 0x60, 0xbd, 0x0d, 0x65, 0x3a, 0x5a, 0x34, 0x33, 0xf6, 0xda, 0x40, 0x43, - 0x1a, 0x6b, 0xce, 0xf5, 0x39, 0xf7, 0x75, 0x2e, 0xf4, 0xfd, 0xc8, 0x5d, 0xc4, 0x11, 0x8f, 0x2c, - 0xf9, 0xc5, 0xed, 0x02, 0x0f, 0x86, 0x6e, 0x14, 0xb9, 0x3e, 0x9b, 0x48, 0x34, 0x4f, 0xdf, 0x4f, - 0xb8, 0x17, 0xb0, 0x84, 0x3b, 0xc1, 0x42, 0x51, 0x07, 0xb7, 0x5c, 0x8f, 0x1f, 0xa5, 0x73, 0xeb, - 0x20, 0x0a, 0x26, 0x6e, 0xe4, 0x46, 0x25, 0x53, 0x20, 0x09, 0xe4, 0x4b, 0xd1, 0xcd, 0x87, 0xd0, - 0x79, 0x95, 0x26, 0x47, 0x94, 0x1d, 0xa7, 0x2c, 0xe1, 0x78, 0x07, 0x5a, 0x09, 0x8f, 0x99, 0x13, - 0x24, 0x06, 0x1a, 0xd5, 0xc6, 0x9d, 0xe9, 0x96, 0xb5, 0x69, 0xe5, 0xb5, 0xfc, 0x41, 0x0b, 0x82, - 0xd9, 0x87, 0xae, 0x92, 0x26, 0x8b, 0x28, 0x4c, 0x98, 0xf9, 0x13, 0x41, 0x77, 0x3f, 0x65, 0xf1, - 0x49, 0x91, 0xec, 0x2a, 0x34, 0x8e, 0x05, 0x36, 0xd0, 0x08, 0x8d, 0x75, 0xaa, 0x80, 0x88, 0xfa, - 0x5e, 0xe0, 0x71, 0xe3, 0xbf, 0x11, 0x1a, 0xf7, 0xa8, 0x02, 0x78, 0x17, 0x1a, 0x09, 0x77, 0x62, - 0x6e, 0xd4, 0x46, 0x68, 0xdc, 0x99, 0x0e, 0x2c, 0x35, 0xa7, 0x55, 0x74, 0x6f, 0xbd, 0x29, 0xe6, - 0xb4, 0xdb, 0x67, 0x5f, 0x87, 0xda, 0xe9, 0xb7, 0x21, 0xa2, 0x4a, 0x82, 0xef, 0x43, 0x8d, 0x85, - 0x87, 0x46, 0xfd, 0x12, 0x4a, 0x21, 0xc0, 0x77, 0x40, 0x3f, 0xf4, 0x62, 0x76, 0xc0, 0xbd, 0x28, - 0x34, 0x1a, 0x23, 0x34, 0xee, 0x4f, 0xaf, 0x94, 0xe3, 0xee, 0x15, 0xbf, 0x68, 0xc9, 0x12, 0xcd, - 0xc7, 0xcc, 0x65, 0x1f, 0x8d, 0xa6, 0x1a, 0x49, 0x02, 0xf3, 0x11, 0xf4, 0xf2, 0xc1, 0xd5, 0x2a, - 0x2e, 0xb5, 0xc6, 0x5d, 0xe8, 0xbe, 0x70, 0xe6, 0xcc, 0x2f, 0xb6, 0x86, 0xa1, 0x1e, 0x3a, 0x01, - 0xcb, 0x97, 0x26, 0xdf, 0x78, 0x1b, 0x9a, 0x1f, 0x1c, 0x3f, 0x65, 0x89, 0x5c, 0x5a, 0x9b, 0xe6, - 0xc8, 0xbc, 0x09, 0xbd, 0x5c, 0x9b, 0x17, 0x2e, 0x89, 0xa2, 0xae, 0xbe, 0x21, 0xee, 0x43, 0x53, - 0xd5, 0x15, 0x0c, 0x5f, 0x48, 0x92, 0xbc, 0x40, 0x8e, 0xf0, 0x04, 0x5a, 0x2c, 0xe4, 0xb1, 0x27, - 0x6b, 0x88, 0x96, 0xff, 0x2f, 0x5b, 0x7e, 0x1a, 0xf2, 0xf8, 0xc4, 0xae, 0x8b, 0xed, 0xd1, 0x82, - 0x65, 0xbe, 0x83, 0x86, 0x8c, 0x63, 0x1b, 0xf4, 0xcd, 0x11, 0xca, 0xa4, 0xff, 0x6a, 0x42, 0x29, - 0x13, 0x43, 0xfb, 0x5e, 0xc8, 0xe4, 0x78, 0x3a, 0x95, 0xef, 0x9d, 0x1b, 0xa0, 0x6f, 0x3c, 0xc0, - 0x1d, 0x68, 0x3d, 0x7b, 0x49, 0xdf, 0xce, 0xe8, 0xde, 0x96, 0x86, 0xbb, 0xd0, 0xb6, 0x67, 0x4f, - 0x9e, 0x4b, 0x84, 0xa6, 0x33, 0x68, 0x8a, 0x3b, 0x64, 0x31, 0x7e, 0x00, 0x75, 0xf1, 0xc2, 0xd7, - 0xca, 0xd6, 0x2b, 0xc7, 0x3d, 0xd8, 0xfe, 0x33, 0x9c, 0x1f, 0xae, 0x36, 0xfd, 0x8c, 0xa0, 0x25, - 0x1c, 0xf4, 0x58, 0x8c, 0x1f, 0x43, 0x43, 0x9a, 0x89, 0x2b, 0xf4, 0xea, 0x59, 0x0f, 0xae, 0xff, - 0x15, 0x2f, 0xf2, 0xdc, 0x46, 0xe2, 0x8e, 0xa5, 0x23, 0x55, 0x75, 0xd5, 0xde, 0xaa, 0xfa, 0x37, - 0xeb, 0x4c, 0xcd, 0xbe, 0xb7, 0x5c, 0x11, 0xed, 0x7c, 0x45, 0xb4, 0x8b, 0x15, 0x41, 0x9f, 0x32, - 0x82, 0xbe, 0x64, 0x04, 0x9d, 0x65, 0x04, 0x2d, 0x33, 0x82, 0xbe, 0x67, 0x04, 0xfd, 0xc8, 0x88, - 0x76, 0x91, 0x11, 0x74, 0xba, 0x26, 0xda, 0x72, 0x4d, 0xb4, 0xf3, 0x35, 0xd1, 0xe6, 0x4d, 0x99, - 0xec, 0xee, 0xaf, 0x00, 0x00, 0x00, 0xff, 0xff, 0xec, 0x0b, 0x27, 0xf0, 0x34, 0x04, 0x00, 0x00, +func init() { proto.RegisterFile("logproto.proto", fileDescriptor_logproto_5d6f332523c5fa62) } + +var fileDescriptor_logproto_5d6f332523c5fa62 = []byte{ + // 592 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x53, 0x4d, 0x4f, 0xd4, 0x40, + 0x18, 0xee, 0xc0, 0x7e, 0xbe, 0xbb, 0x20, 0x19, 0x15, 0x9b, 0x3d, 0x0c, 0x9b, 0x1e, 0x74, 0x43, + 0x62, 0x57, 0x57, 0xa3, 0x11, 0xb9, 0xb0, 0x22, 0x17, 0x4d, 0x94, 0xd1, 0xc4, 0xa3, 0xe9, 0xc2, + 0x50, 0x9a, 0xf4, 0x63, 0x99, 0x99, 0x92, 0x70, 0x33, 0xf1, 0x6c, 0xc2, 0xcf, 0xf0, 0xa7, 0x70, + 0xe4, 0xc8, 0x49, 0xa5, 0x7b, 0xf1, 0xc8, 0x4f, 0x30, 0x33, 0xed, 0x6c, 0x2b, 0x5e, 0xe4, 0xd2, + 0xcc, 0xf3, 0xce, 0xf3, 0xcc, 0xfb, 0xf1, 0xf4, 0x85, 0xe5, 0x30, 0xf1, 0xa7, 0x3c, 0x91, 0x89, + 0xab, 0xbf, 0xb8, 0x65, 0x70, 0x6f, 0xcd, 0x4f, 0x12, 0x3f, 0x64, 0x43, 0x8d, 0x26, 0xe9, 0xc1, + 0x50, 0x06, 0x11, 0x13, 0xd2, 0x8b, 0xa6, 0x39, 0xb5, 0xf7, 0xd0, 0x0f, 0xe4, 0x61, 0x3a, 0x71, + 0xf7, 0x92, 0x68, 0xe8, 0x27, 0x7e, 0x52, 0x32, 0x15, 0xd2, 0x40, 0x9f, 0x72, 0xba, 0xf3, 0x02, + 0x3a, 0xef, 0x53, 0x71, 0x48, 0xd9, 0x51, 0xca, 0x84, 0xc4, 0xeb, 0xd0, 0x14, 0x92, 0x33, 0x2f, + 0x12, 0x36, 0xea, 0x2f, 0x0e, 0x3a, 0xa3, 0x15, 0x77, 0x5e, 0xca, 0x07, 0x7d, 0x41, 0x0d, 0xc1, + 0x59, 0x86, 0x6e, 0x2e, 0x15, 0xd3, 0x24, 0x16, 0xcc, 0xf9, 0xb6, 0x00, 0xdd, 0xdd, 0x94, 0xf1, + 0x13, 0xf3, 0xd8, 0x1d, 0xa8, 0x1f, 0x29, 0x6c, 0xa3, 0x3e, 0x1a, 0xb4, 0x69, 0x0e, 0x54, 0x34, + 0x0c, 0xa2, 0x40, 0xda, 0x0b, 0x7d, 0x34, 0x58, 0xa2, 0x39, 0xc0, 0x1b, 0x50, 0x17, 0xd2, 0xe3, + 0xd2, 0x5e, 0xec, 0xa3, 0x41, 0x67, 0xd4, 0x73, 0xf3, 0x3e, 0x5d, 0x53, 0xbd, 0xfb, 0xd1, 0xf4, + 0x39, 0x6e, 0x9d, 0xfd, 0x58, 0xb3, 0x4e, 0x7f, 0xae, 0x21, 0x9a, 0x4b, 0xf0, 0x33, 0x58, 0x64, + 0xf1, 0xbe, 0x5d, 0xbb, 0x81, 0x52, 0x09, 0xf0, 0x63, 0x68, 0xef, 0x07, 0x9c, 0xed, 0xc9, 0x20, + 0x89, 0xed, 0x7a, 0x1f, 0x0d, 0x96, 0x47, 0xb7, 0xcb, 0x76, 0xb7, 0xcd, 0x15, 0x2d, 0x59, 0x6a, + 0x3e, 0x07, 0x41, 0x28, 0x19, 0x17, 0x76, 0xe3, 0xfa, 0x7c, 0x76, 0xf4, 0x05, 0x35, 0x04, 0x67, + 0x13, 0x1a, 0x79, 0x08, 0xdb, 0xd0, 0x9c, 0x7a, 0x52, 0x32, 0x1e, 0x17, 0xa3, 0x30, 0x50, 0xdd, + 0x04, 0xf1, 0x31, 0xe3, 0x82, 0xe9, 0x71, 0xb4, 0xa8, 0x81, 0xce, 0x4b, 0x58, 0x2a, 0x86, 0x99, + 0x8f, 0xf7, 0x46, 0xd6, 0x6c, 0x40, 0xf7, 0xad, 0x37, 0x61, 0xa1, 0x71, 0x02, 0x43, 0x2d, 0xf6, + 0x22, 0x56, 0x64, 0xd7, 0x67, 0xbc, 0x0a, 0x8d, 0x63, 0x2f, 0x4c, 0x99, 0x28, 0x32, 0x17, 0xc8, + 0x79, 0x00, 0x4b, 0x85, 0xb6, 0x48, 0x5c, 0x12, 0x55, 0xde, 0xf6, 0x9c, 0xb8, 0x0b, 0x8d, 0x3c, + 0xaf, 0x62, 0x84, 0x4a, 0x22, 0x8a, 0x04, 0x05, 0xc2, 0x43, 0x68, 0xb2, 0x58, 0xf2, 0x40, 0xe7, + 0x50, 0x25, 0xdf, 0x2a, 0x4b, 0x7e, 0x1d, 0x4b, 0x7e, 0x32, 0xae, 0x29, 0x47, 0xa8, 0x61, 0x39, + 0x9f, 0xa1, 0xae, 0xe3, 0x78, 0x0c, 0xed, 0xf9, 0x8f, 0xad, 0x1f, 0xfd, 0x5f, 0x63, 0x4b, 0x99, + 0x6a, 0x3a, 0x0c, 0xe2, 0x7c, 0xb0, 0x6d, 0xaa, 0xcf, 0xeb, 0xf7, 0xa1, 0x3d, 0xf7, 0x15, 0x77, + 0xa0, 0xb9, 0xf3, 0x8e, 0x7e, 0xda, 0xa2, 0xdb, 0x2b, 0x16, 0xee, 0x42, 0x6b, 0xbc, 0xf5, 0xea, + 0x8d, 0x46, 0x68, 0xb4, 0x05, 0x0d, 0xf5, 0x6f, 0x33, 0x8e, 0x9f, 0x43, 0x4d, 0x9d, 0xf0, 0xdd, + 0xb2, 0xf4, 0xca, 0xc2, 0xf4, 0x56, 0xaf, 0x87, 0x8b, 0x65, 0xb0, 0x46, 0x5f, 0x11, 0x34, 0x95, + 0x83, 0x01, 0xe3, 0x78, 0x13, 0xea, 0xda, 0x4c, 0x5c, 0xa1, 0x57, 0x57, 0xa5, 0x77, 0xef, 0x9f, + 0xb8, 0x79, 0xe7, 0x11, 0x52, 0xbb, 0xa1, 0x1d, 0xa9, 0xaa, 0xab, 0xf6, 0x56, 0xd5, 0x7f, 0x59, + 0xe7, 0x58, 0xe3, 0xa7, 0xe7, 0x97, 0xc4, 0xba, 0xb8, 0x24, 0xd6, 0xd5, 0x25, 0x41, 0x5f, 0x32, + 0x82, 0xbe, 0x67, 0x04, 0x9d, 0x65, 0x04, 0x9d, 0x67, 0x04, 0xfd, 0xca, 0x08, 0xfa, 0x9d, 0x11, + 0xeb, 0x2a, 0x23, 0xe8, 0x74, 0x46, 0xac, 0xf3, 0x19, 0xb1, 0x2e, 0x66, 0xc4, 0x9a, 0x34, 0xf4, + 0x63, 0x4f, 0xfe, 0x04, 0x00, 0x00, 0xff, 0xff, 0x98, 0xa6, 0xbf, 0xde, 0x88, 0x04, 0x00, 0x00, } diff --git a/pkg/logproto/logproto.proto b/pkg/logproto/logproto.proto index 8a648176..1a88f705 100644 --- a/pkg/logproto/logproto.proto +++ b/pkg/logproto/logproto.proto @@ -27,7 +27,13 @@ message QueryRequest { google.protobuf.Timestamp start = 3 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false]; google.protobuf.Timestamp end = 4 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false]; Direction direction = 5; - string regex = 6; + + repeated Filter filters = 6; +} + +message Filter { + string pattern = 1; + bool inverse = 2; } enum Direction { diff --git a/pkg/querier/http.go b/pkg/querier/http.go index 4a1f218d..738c3354 100644 --- a/pkg/querier/http.go +++ b/pkg/querier/http.go @@ -92,7 +92,7 @@ func (q *Querier) QueryHandler(w http.ResponseWriter, r *http.Request) { Start: start, End: end, Direction: direction, - Regex: params.Get("regexp"), + Filters: makeFilters(params["regexp"], params["invregexp"]), } log.Printf("Query request: %+v", request) @@ -108,6 +108,25 @@ func (q *Querier) QueryHandler(w http.ResponseWriter, r *http.Request) { } } +func makeFilters(filterStrs, inverseFilterStrs []string) []*logproto.Filter { + filters := make([]*logproto.Filter, 0, len(filterStrs)) + + for _, filterStr := range filterStrs { + filters = append(filters, &logproto.Filter{ + Pattern: filterStr, + }) + } + + for _, filterStr := range inverseFilterStrs { + filters = append(filters, &logproto.Filter{ + Pattern: filterStr, + Inverse: true, + }) + } + + return filters +} + // LabelHandler is a http.HandlerFunc for handling label queries. func (q *Querier) LabelHandler(w http.ResponseWriter, r *http.Request) { name, ok := mux.Vars(r)["name"] -- GitLab