| 3856 | } |
| 3857 | |
| 3858 | func (i *mockIngester) QueryStream(ctx context.Context, req *client.QueryRequest, opts ...grpc.CallOption) (client.Ingester_QueryStreamClient, error) { |
| 3859 | time.Sleep(i.queryDelay) |
| 3860 | |
| 3861 | i.Lock() |
| 3862 | defer i.Unlock() |
| 3863 | |
| 3864 | i.trackCall("QueryStream") |
| 3865 | |
| 3866 | if !i.happy.Load() { |
| 3867 | return nil, errFail |
| 3868 | } |
| 3869 | |
| 3870 | _, _, matchers, err := client.FromQueryRequest(storecache.NoopMatchersCache, req) |
| 3871 | if err != nil { |
| 3872 | return nil, err |
| 3873 | } |
| 3874 | |
| 3875 | results := []*client.QueryStreamResponse{} |
| 3876 | for _, ts := range i.timeseries { |
| 3877 | if !match(ts.Labels, matchers) { |
| 3878 | continue |
| 3879 | } |
| 3880 | |
| 3881 | c := chunkenc.NewXORChunk() |
| 3882 | appender, err := c.Appender() |
| 3883 | if err != nil { |
| 3884 | return nil, err |
| 3885 | } |
| 3886 | chunks := []chunkenc.Chunk{c} |
| 3887 | for _, sample := range ts.Samples { |
| 3888 | appender.Append(sample.TimestampMs, sample.Value) |
| 3889 | } |
| 3890 | |
| 3891 | wireChunks := []client.Chunk{} |
| 3892 | for _, c := range chunks { |
| 3893 | e, err := promchunk.FromPromChunkEncoding(c.Encoding()) |
| 3894 | if err != nil { |
| 3895 | return nil, err |
| 3896 | } |
| 3897 | chunk := client.Chunk{ |
| 3898 | Encoding: int32(e), |
| 3899 | Data: c.Bytes(), |
| 3900 | } |
| 3901 | wireChunks = append(wireChunks, chunk) |
| 3902 | } |
| 3903 | |
| 3904 | results = append(results, &client.QueryStreamResponse{ |
| 3905 | Chunkseries: []client.TimeSeriesChunk{ |
| 3906 | { |
| 3907 | Labels: ts.Labels, |
| 3908 | Chunks: wireChunks, |
| 3909 | }, |
| 3910 | }, |
| 3911 | }) |
| 3912 | } |
| 3913 | return &queryStream{ |
| 3914 | results: results, |
| 3915 | }, nil |