Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions configs/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ go_library(
"arista_default.textproto",
"lemming_default.textproto",
"nokia_default.textproto",
"fault_config.textproto",
],
importpath = "github.com/openconfig/lemming/configs",
visibility = ["//visibility:public"],
Expand Down
19 changes: 19 additions & 0 deletions configs/fault_config.textproto
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
fault_config {
gnoi_faults {
rpc_method: "/gnoi.system.System/Reboot"
faults {
msg_id: "reboot_fault_1"
status {
code: 10
message: "Simulated reboot failure - system maintenance mode"
}
}
faults {
msg_id: "reboot_fault_2"
status {
code: 14
message: "Simulated reboot failure - resource temporarily unavailable"
}
}
}
}
9 changes: 8 additions & 1 deletion fault/BUILD
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ go_library(
importpath = "github.com/openconfig/lemming/fault",
visibility = ["//visibility:public"],
deps = [
"//proto/config",
"//proto/fault",
"@com_github_golang_glog//:glog",
"@com_github_google_uuid//:uuid",
Expand All @@ -23,13 +24,19 @@ go_library(

go_test(
name = "fault_test",
srcs = ["fault_test.go"],
srcs = [
"fault_test.go",
"interceptor_test.go",
],
embed = [":fault"],
deps = [
"//fault/proto/test",
"//proto/config",
"//proto/fault",
"@com_github_google_go_cmp//cmp",
"@com_github_openconfig_gnmi//errdiff",
"@com_github_openconfig_gnoi//system",
"@org_golang_google_genproto_googleapis_rpc//status",
"@org_golang_google_grpc//:grpc",
"@org_golang_google_grpc//codes",
"@org_golang_google_grpc//credentials/insecure",
Expand Down
96 changes: 96 additions & 0 deletions fault/config_fault/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,96 @@
# Config-Based Fault Injection

## Overview

Config-based fault injection allows you to pre-configure fault responses for specific gNOI RPC methods directly in the Lemming configuration file. This provides deterministic, reproducible fault scenarios for integration testing without requiring external fault injection clients.

## Configuration

### Basic Structure

```protobuf
fault_config {
gnoi_faults {
rpc_method: "/gnoi.system.System/Reboot"
faults {
msg_id: "reboot_fault_1"
status {
code: 3 # INVALID_ARGUMENT
message: "Simulated reboot failure"
}
}
}
}
```

### Configuration Fields

- **`rpc_method`**: Full gRPC method name (e.g., `/gnoi.system.System/Reboot`)
- **`faults`**: Array of fault messages to inject
- **`msg_id`**: Unique identifier for the fault
- **`msg`**: Optional modified request/response message
- **`status`**: gRPC status to return (code and message)

## Fault Behavior

### Exhaustible Application

When multiple faults are configured for an RPC method, they are applied sequentially and then exhausted:

```protobuf
gnoi_faults {
rpc_method: "/gnoi.system.System/Reboot"
faults {
msg_id: "fault_1"
status { code: 3 message: "First fault" }
}
faults {
msg_id: "fault_2"
status { code: 14 message: "Second fault" }
}
}
```

- 1st call → fault_1 (INVALID_ARGUMENT)
- 2nd call → fault_2 (UNAVAILABLE)
- 3rd call → **normal behavior** (continues normally)

### Fault Types

#### Status-Only Faults
Return an error without executing the RPC:

```protobuf
faults {
msg_id: "permission_denied"
status {
code: 7 # PERMISSION_DENIED
message: "Access denied in maintenance mode"
}
}
```

#### Message Modification Faults
Modify the request before processing:

```protobuf
faults {
msg_id: "modify_request"
msg {
# Encoded modified request message
type_url: "type.googleapis.com/gnoi.system.RebootRequest"
value: "..." # Modified request proto bytes
}
status {
code: 0 # OK - process with modified request
}
}
```

### Running Lemming with Fault Config

```bash
# Start lemming with fault configuration
./lemming \
--config_file=path_to_fault_config
```
121 changes: 117 additions & 4 deletions fault/fault.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,22 +29,43 @@ import (

"github.com/google/uuid"

configpb "github.com/openconfig/lemming/proto/config"
faultpb "github.com/openconfig/lemming/proto/fault"
)

func NewInterceptor() *Interceptor {
return &Interceptor{
receivers: map[string]chan *faultMessage{},
faultSubs: map[string]*faultSubscription{},
receivers: map[string]chan *faultMessage{},
faultSubs: map[string]*faultSubscription{},
configuredFaults: map[string][]*faultpb.FaultMessage{},
}
}

// NewInterceptorFromConfig creates a new interceptor configured with static faults from config
func NewInterceptorFromConfig(faultConfig *configpb.FaultServiceConfiguration) *Interceptor {
interceptor := NewInterceptor()

if faultConfig != nil {
for _, gnoiFault := range faultConfig.GetGnoiFaults() {
if err := interceptor.configureFaults(gnoiFault.GetRpcMethod(), gnoiFault.GetFaults()); err != nil {
log.Warningf("Failed to configure faults for RPC method %s: %v", gnoiFault.GetRpcMethod(), err)
}
}
}

return interceptor
}

type Interceptor struct {
faultpb.UnimplementedFaultInjectServer
faultSubsMu sync.Mutex
faultSubs map[string]*faultSubscription
receiversMu sync.Mutex
receivers map[string]chan *faultMessage
// configuredFaults is a map of RPC methods to a slice of faults.
configuredFaults map[string][]*faultpb.FaultMessage
// Mutex for configured faults.
configMu sync.Mutex
}

type faultMessage struct {
Expand Down Expand Up @@ -117,8 +138,90 @@ func (i *Interceptor) sendRecvFault(ch chan *faultMessage, rpcID string, msg any
return res, recv.status.Err()
}

// Unary implements the grpc unary server imterceptor interface, adding fault injection to unary RPCs.
// configureFaults sets pre-configured faults for a specific RPC method
func (i *Interceptor) configureFaults(rpcMethod string, faults []*faultpb.FaultMessage) error {
i.configMu.Lock()
defer i.configMu.Unlock()

if rpcMethod == "" {
Comment thread
mtr002 marked this conversation as resolved.
return status.Errorf(codes.InvalidArgument, "rpc_method cannot be empty")
}

if len(faults) == 0 {
delete(i.configuredFaults, rpcMethod)
log.Infof("Removed fault configuration for RPC method: %s", rpcMethod)
return nil
}

i.configuredFaults[rpcMethod] = faults
log.Infof("Configured %d faults for RPC method: %s", len(faults), rpcMethod)
return nil
}

// nextConfiguredFault returns the next configured fault for an RPC method
func (i *Interceptor) nextConfiguredFault(rpcMethod string) *faultpb.FaultMessage {
Comment thread
mtr002 marked this conversation as resolved.
i.configMu.Lock()
defer i.configMu.Unlock()

faults, exists := i.configuredFaults[rpcMethod]
if !exists || len(faults) == 0 {
return nil
}

// Consume the first fault from the slice
fault := faults[0]
i.configuredFaults[rpcMethod] = faults[1:]

return fault
}

// applyConfiguredFault applies a configured fault to the message
func (i *Interceptor) applyConfiguredFault(rpcMethod string, fault *faultpb.FaultMessage, originalMsg any, originalErr error) (any, error) {
if fault == nil {
return originalMsg, originalErr
}

log.Infof("Applying configured fault for RPC %s: msg_id=%s", rpcMethod, fault.GetMsgId())

// If fault has a message, use it; otherwise use original.
if fault.GetMsg() != nil {
faultMsg, err := fault.GetMsg().UnmarshalNew()
if err == nil {
if fault.GetStatus() != nil {
// If status is OK, return modified message with no error
if fault.GetStatus().Code == 0 {
return faultMsg, nil
}
// Otherwise return the fault status error
return faultMsg, status.FromProto(fault.GetStatus()).Err()
}
// If no status provided, return modified message with default error
return faultMsg, status.Errorf(codes.Internal, "configured fault without status for %s", rpcMethod)
}
log.Warningf("Failed to unmarshal fault message for %s: %v", rpcMethod, err)
}

// If fault only has status, return original message with fault status
if fault.GetStatus() != nil {
return originalMsg, status.FromProto(fault.GetStatus()).Err()
}
return originalMsg, status.Errorf(codes.Internal, "configured fault without status for %s", rpcMethod)
}

// Unary implements the grpc unary server interceptor interface, adding fault injection to unary RPCs.
func (i *Interceptor) Unary(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
// Check for configured faults first
if fault := i.nextConfiguredFault(info.FullMethod); fault != nil {
modifiedReq, faultErr := i.applyConfiguredFault(info.FullMethod, fault, req, nil)

// If fault has an error status, return immediately without calling handler.
if faultErr != nil {
return modifiedReq, faultErr
}
req = modifiedReq
}

// Check for live fault subscriptions
i.faultSubsMu.Lock()
sub, ok := i.faultSubs[info.FullMethod]
i.faultSubsMu.Unlock()
Expand Down Expand Up @@ -165,12 +268,22 @@ func (si *streamInt) SendMsg(m any) error {
return si.ServerStream.SendMsg(msg)
}

// Stream implements the grpc strean server imterceptor interface, adding fault injection to streanubg RPCs.
// Stream implements the grpc stream server interceptor interface, adding fault injection to streaming RPCs.
func (i *Interceptor) Stream(srv any, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
if info.FullMethod == "/lemming.fault.FaultInject/Intercept" { // Do not self-intercept
return handler(srv, stream)
}

// Check for configured faults first - for streaming, apply fault immediately
if fault := i.nextConfiguredFault(info.FullMethod); fault != nil {
log.Infof("Applying configured stream fault for RPC %s: msg_id=%s", info.FullMethod, fault.GetMsgId())
if fault.GetStatus() != nil {
return status.FromProto(fault.GetStatus()).Err()
}
return status.Errorf(codes.Internal, "configured fault without status for %s", info.FullMethod)
}

// Check for live fault subscriptions
i.faultSubsMu.Lock()
sub, ok := i.faultSubs[info.FullMethod]
i.faultSubsMu.Unlock()
Expand Down
Loading
Loading