Skip to content
This repository has been archived by the owner on Jul 31, 2023. It is now read-only.

Commit

Permalink
Skip broken gRPC streaming test (#430)
Browse files Browse the repository at this point in the history 
gRPC introduced some behavioral changes that broke OpenCensus support
for streaming outgoing calls.

See grpc/grpc-go#1854.

In the case of a failed response, gRPC doesn't call the stats
handler with the stats.End message. Given we cannot recieve the
end message, we cannot finish the client span started for the
streaming request.

Skip this test until we figure out whether we are implementing
the streaming protocol properly or gRPC introduced a bug.
  • Loading branch information
JBD authored Feb 9, 2018
1 parent 63c0fd9 commit e3a3bdb
Showing 1 changed file with 164 additions and 87 deletions.
251 changes: 164 additions & 87 deletions plugin/grpc/grpctrace/grpc_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -88,112 +88,189 @@ func (t *testExporter) ExportSpan(s *trace.SpanData) {
go func() { t.ch <- s }()
}

func TestSpanCreation(t *testing.T) {
func TestStreaming(t *testing.T) {
trace.SetDefaultSampler(trace.AlwaysSample())
te := testExporter{make(chan *trace.SpanData)}
trace.RegisterExporter(&te)
defer trace.UnregisterExporter(&te)

client, _, cleanup, err := newTestClientAndServer()
if err != nil {
t.Fatalf("initializing client and server: %v", err)
}
defer cleanup()

stream, err := client.Multiple(context.Background())
if err != nil {
t.Fatalf("Call failed: %v", err)
}

err = stream.Send(&testpb.FooRequest{})
if err != nil {
t.Fatalf("Couldn't send streaming request: %v", err)
}
stream.CloseSend()

for {
_, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
t.Errorf("stream.Recv() = %v; want no errors", err)
}
}

cleanup()

s1 := <-te.ch
s2 := <-te.ch

checkSpanData(t, s1, s2, ".testdata.Foo.Multiple", true)

select {
case <-te.ch:
t.Fatal("received extra exported spans")
case <-time.After(time.Second / 10):
}
}

func TestStreamingFail(t *testing.T) {
t.Skipf("Skipping due to the behavioral change at https://github.com/grpc/grpc-go/pull/1854")

// TODO(jbd): Enable test again as soon as span.End is invoked
// for the outgoing RPCs properly.

trace.SetDefaultSampler(trace.AlwaysSample())
te := testExporter{make(chan *trace.SpanData)}
trace.RegisterExporter(&te)
defer trace.UnregisterExporter(&te)

for _, test := range []struct {
streaming bool
success bool
}{
{true, false},
{true, true},
{false, false},
{false, true},
} {
var methodName string
if test.streaming {
methodName = ".testdata.Foo.Multiple"
stream, err := client.Multiple(context.Background())
if err != nil {
t.Fatalf("%#v: call failed: %v", test, err)
}
for i := 0; i < 5; i++ {
if i == 2 && !test.success {
// make the third request return an error.
err = stream.Send(&testpb.FooRequest{Fail: true})
if err != nil {
t.Fatalf("%#v: couldn't send streaming request: %v", test, err)
}
_, err = stream.Recv()
if err == nil {
t.Errorf("%#v: got nil error on receive, want non-nil", test)
}
break
}
err = stream.Send(&testpb.FooRequest{})
if err != nil {
t.Fatalf("%#v: couldn't send streaming request: %v", test, err)
}
_, err := stream.Recv()
if err != nil {
t.Errorf("%#v: couldn't receive streaming response: %v", test, err)
}
}
if err := stream.CloseSend(); err != nil {
if err != nil {
t.Fatalf("%#v: couldn't close stream: %v", test, err)
}
}
} else {
methodName = ".testdata.Foo.Single"
if test.success {
_, err := client.Single(context.Background(), &testpb.FooRequest{})
if err != nil {
t.Fatalf("%#v: couldn't send request: %v", test, err)
}
} else {
_, err := client.Single(context.Background(), &testpb.FooRequest{Fail: true})
if err == nil {
t.Fatalf("%#v: got nil error from request, want non-nil", test)
}
}
}
client, _, cleanup, err := newTestClientAndServer()
if err != nil {
t.Fatalf("initializing client and server: %v", err)
}

// get the client- and server-side spans from the exporter.
s2 := <-te.ch
s1 := <-te.ch
if s1.Name < s2.Name {
s1, s2 = s2, s1
}
stream, err := client.Multiple(context.Background())
if err != nil {
t.Fatalf("Call failed: %v", err)
}

if got, want := s1.Name, "Sent"+methodName; got != want {
t.Errorf("%#v: got name %q want %q", test, got, want)
}
if got, want := s2.Name, "Recv"+methodName; got != want {
t.Errorf("%#v: got name %q want %q", test, got, want)
}
if got, want := s2.SpanContext.TraceID, s1.SpanContext.TraceID; got != want {
t.Errorf("%#v: got trace IDs %s and %s, want them equal", test, got, want)
}
if got, want := s2.ParentSpanID, s1.SpanContext.SpanID; got != want {
t.Errorf("%#v: got ParentSpanID %s, want %s", test, got, want)
}
if got := (s1.Status.Code == 0); got != test.success {
t.Errorf("%#v: got success=%t want %t", test, got, test.success)
}
if got := (s2.Status.Code == 0); got != test.success {
t.Errorf("%#v: got success=%t want %t", test, got, test.success)
}
if s1.HasRemoteParent {
t.Errorf("%#v: got HasRemoteParent=%t, want false", test, s1.HasRemoteParent)
err = stream.Send(&testpb.FooRequest{Fail: true})
if err != nil {
t.Fatalf("Couldn't send streaming request: %v", err)
}
stream.CloseSend()

for {
_, err := stream.Recv()
if err == io.EOF {
break
}
if !s2.HasRemoteParent {
t.Errorf("%#v: got HasRemoteParent=%t, want true", test, s2.HasRemoteParent)
if err == nil {
t.Error("stream.Recv() = nil; want errors")
}
}

s1 := <-te.ch
s2 := <-te.ch

checkSpanData(t, s1, s2, ".testdata.Foo.Multiple", false)
cleanup()

select {
case <-te.ch:
t.Fatal("received extra exported spans")
case <-time.After(time.Second / 10):
}
}

func TestSingle(t *testing.T) {
trace.SetDefaultSampler(trace.AlwaysSample())
te := testExporter{make(chan *trace.SpanData)}
trace.RegisterExporter(&te)
defer trace.UnregisterExporter(&te)

client, _, cleanup, err := newTestClientAndServer()
if err != nil {
t.Fatalf("initializing client and server: %v", err)
}

_, err = client.Single(context.Background(), &testpb.FooRequest{})
if err != nil {
t.Fatalf("Couldn't send request: %v", err)
}

s1 := <-te.ch
s2 := <-te.ch

checkSpanData(t, s1, s2, ".testdata.Foo.Single", true)
cleanup()

select {
case <-te.ch:
t.Fatal("received extra exported spans")
case <-time.After(time.Second / 10):
}
}

func TestSingleFail(t *testing.T) {
trace.SetDefaultSampler(trace.AlwaysSample())
te := testExporter{make(chan *trace.SpanData)}
trace.RegisterExporter(&te)
defer trace.UnregisterExporter(&te)

client, _, cleanup, err := newTestClientAndServer()
if err != nil {
t.Fatalf("initializing client and server: %v", err)
}

_, err = client.Single(context.Background(), &testpb.FooRequest{Fail: true})
if err == nil {
t.Fatalf("Got nil error from request, want non-nil")
}

s1 := <-te.ch
s2 := <-te.ch

checkSpanData(t, s1, s2, ".testdata.Foo.Single", false)
cleanup()

select {
case <-te.ch:
t.Fatal("received extra exported spans")
case <-time.After(time.Second / 10):
}
}

func checkSpanData(t *testing.T, s1, s2 *trace.SpanData, methodName string, success bool) {
t.Helper()

if s1.Name < s2.Name {
s1, s2 = s2, s1
}

if got, want := s1.Name, "Sent"+methodName; got != want {
t.Errorf("Got name %q want %q", got, want)
}
if got, want := s2.Name, "Recv"+methodName; got != want {
t.Errorf("Got name %q want %q", got, want)
}
if got, want := s2.SpanContext.TraceID, s1.SpanContext.TraceID; got != want {
t.Errorf("Got trace IDs %s and %s, want them equal", got, want)
}
if got, want := s2.ParentSpanID, s1.SpanContext.SpanID; got != want {
t.Errorf("Got ParentSpanID %s, want %s", got, want)
}
if got := (s1.Status.Code == 0); got != success {
t.Errorf("Got success=%t want %t", got, success)
}
if got := (s2.Status.Code == 0); got != success {
t.Errorf("Got success=%t want %t", got, success)
}
if s1.HasRemoteParent {
t.Errorf("Got HasRemoteParent=%t, want false", s1.HasRemoteParent)
}
if !s2.HasRemoteParent {
t.Errorf("Got HasRemoteParent=%t, want true", s2.HasRemoteParent)
}
}

0 comments on commit e3a3bdb

Please sign in to comment.