CassandraSpanStore implements SpanStore, Traces, ServiceAndSpanNames { @Override public Call<List<List<Span>>> getTraces(QueryRequest request) { if (!searchEnabled) return Call.emptyList(); if (request.minDuration() != null) { throw new IllegalArgumentException( "getTraces with duration is unsupported. Upgrade to cassandra3."); } final int traceIndexFetchSize = request.limit() * indexFetchMultiplier; List<Call<Set<Pair>>> callsToIntersect = new ArrayList<>(); List<String> annotationKeys = CassandraUtil.annotationKeys(request); if (request.serviceName() != null) { String remoteService = request.remoteServiceName(); if (request.spanName() != null || remoteService != null) { if (request.spanName() != null) { callsToIntersect.add( selectTraceIdsBySpanName.newCall( request.serviceName(), request.spanName(), request.endTs() * 1000, request.lookback() * 1000, traceIndexFetchSize)); } if (remoteService != null) { if (selectTraceIdsByRemoteServiceName == null) { throw new IllegalArgumentException("remoteService=" + remoteService + " unsupported due to missing table " + SERVICE_REMOTE_SERVICE_NAME_INDEX); } callsToIntersect.add( selectTraceIdsByRemoteServiceName.newCall( request.serviceName(), remoteService, request.endTs() * 1000, request.lookback() * 1000, traceIndexFetchSize)); } } else { callsToIntersect.add( selectTraceIdsByServiceName.newCall( request.serviceName(), request.endTs() * 1000, request.lookback() * 1000, traceIndexFetchSize)); } for (String annotationKey : annotationKeys) { callsToIntersect.add( selectTraceIdsByAnnotation.newCall( annotationKey, request.endTs() * 1000, request.lookback() * 1000, traceIndexFetchSize)); } } else if (selectTraceIdsByServiceNames == null) { throw new IllegalArgumentException( "getTraces without serviceName requires Cassandra 2.2 or later"); } else { if (!annotationKeys.isEmpty() || request.remoteServiceName() != null || request.spanName() != null) { throw new IllegalArgumentException( "getTraces without serviceName supports no other qualifiers. Upgrade to cassandra3."); } FlatMapper<List<String>, Set<Pair>> flatMapper = selectTraceIdsByServiceNames.newFlatMapper( request.endTs() * 1000, request.lookback() * 1000, traceIndexFetchSize); callsToIntersect.add(getServiceNames().flatMap(flatMapper)); } Call<Set<Long>> traceIdCall; if (callsToIntersect.size() == 1) { traceIdCall = callsToIntersect.get(0).map(sortTraceIdsByDescTimestampMapper()); } else { traceIdCall = new IntersectTraceIds(callsToIntersect); } return traceIdCall.flatMap(spans.newFlatMapper(request)); } CassandraSpanStore(CassandraStorage storage); @Override Call<List<List<Span>>> getTraces(QueryRequest request); @Override Call<List<Span>> getTrace(String traceId); @Override Call<List<List<Span>>> getTraces(Iterable<String> traceIds); @Override Call<List<String>> getServiceNames(); @Override Call<List<String>> getRemoteServiceNames(String serviceName); @Override Call<List<String>> getSpanNames(String serviceName); @Override Call<List<DependencyLink>> getDependencies(long endTs, long lookback);  }
@Test public void getTraces_fansOutAgainstServices() { Call<List<List<Span>>> call = spanStore.getTraces(queryBuilder.build()); assertThat(call.toString()).contains(FlatMapServiceNamesToInput.class.getSimpleName()); }
@Test(expected = IllegalArgumentException.class) public void getTraces_withSpanNameButNoServiceName() { spanStore.getTraces(queryBuilder.spanName("get").build()); }
@Test(expected = IllegalArgumentException.class) public void getTraces_withTagButNoServiceName() { spanStore.getTraces( queryBuilder.annotationQuery(Collections.singletonMap("environment", "production")).build()); }
@Test(expected = IllegalArgumentException.class) public void getTraces_withDurationButNoServiceName() { spanStore.getTraces(queryBuilder.minDuration(100L).build()); }
@Test(expected = IllegalArgumentException.class) public void getTraces_withRemoteServiceNameButNoServiceName() { spanStore.getTraces(queryBuilder.remoteServiceName("backend").build()); }
@Test public void getTraces() { Call<List<List<Span>>> call = spanStore.getTraces(queryBuilder.serviceName("frontend").build()); assertThat(call.toString()).contains("service_name=frontend,"); }
@Test public void getTraces_withSpanName() { Call<List<List<Span>>> call = spanStore.getTraces( queryBuilder.serviceName("frontend").spanName("get").build()); assertThat(call.toString()).contains("service_span_name=frontend.get,"); }
@Test public void getTraces_withRemoteServiceName() { Call<List<List<Span>>> call = spanStore.getTraces( queryBuilder.serviceName("frontend").remoteServiceName("backend").build()); assertThat(call.toString()) .contains("service_remote_service_name=frontend.backend,") .doesNotContain("service=frontend, span="); }
@Test public void getTraces_withSpanNameAndRemoteServiceName() { Call<List<List<Span>>> call = spanStore.getTraces( queryBuilder.serviceName("frontend").remoteServiceName("backend").spanName("get").build()); assertThat(call.toString()) .contains("service_remote_service_name=frontend.backend,") .contains("service_span_name=frontend.get,"); }