gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-11 23:12:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1732 from openziti/zrok-fixes

Browse Source 
zrok fixes
This commit is contained in:
Paul Lorenz
2024-02-08 22:16:39 -05:00
committed by GitHub
parent cd16677213 b9e8a1a903
commit 758bf64da6
36 changed files with 864 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+28
View File
@@ -1,3 +1,31 @@
# Release 0.32.2
## What's New
* Terminator performance improvements
* API Rate Limiter enabled by default
## Component Updates and Bug Fixes
* github.com/openziti/channel/v2: [v2.0.117 -> v2.0.119](https://github.com/openziti/channel/compare/v2.0.117...v2.0.119)
* [Issue #127](https://github.com/openziti/channel/issues/127) - Support some additional types in message headers
* github.com/openziti/foundation/v2: [v2.0.36 -> v2.0.37](https://github.com/openziti/foundation/compare/v2.0.36...v2.0.37)
* github.com/openziti/identity: [v1.0.69 -> v1.0.70](https://github.com/openziti/identity/compare/v1.0.69...v1.0.70)
* github.com/openziti/metrics: [v1.2.43 -> v1.2.45](https://github.com/openziti/metrics/compare/v1.2.43...v1.2.45)
* github.com/openziti/runzmd: [v1.0.37 -> v1.0.38](https://github.com/openziti/runzmd/compare/v1.0.37...v1.0.38)
* github.com/openziti/sdk-golang: [v0.22.21 -> v0.22.28](https://github.com/openziti/sdk-golang/compare/v0.22.21...v0.22.28)
* [Issue #495](https://github.com/openziti/sdk-golang/issues/495) - Refresh edge session if listen attempt fails, to ensure that session is still valid
* github.com/openziti/storage: [v0.2.28 -> v0.2.30](https://github.com/openziti/storage/compare/v0.2.28...v0.2.30)
* github.com/openziti/transport/v2: [v2.0.121 -> v2.0.122](https://github.com/openziti/transport/compare/v2.0.121...v2.0.122)
* github.com/openziti/ziti: [v0.32.1 -> v0.32.2](https://github.com/openziti/ziti/compare/v0.32.1...v0.32.2)
* [Issue #1736](https://github.com/openziti/ziti/issues/1736) - controller crashes generating create circuit responses
* [Issue #1733](https://github.com/openziti/ziti/issues/1733) - Improve terminator creation performance
* [Issue #1734](https://github.com/openziti/ziti/issues/1734) - Make API rate limiter enabled by default
* [Issue #1726](https://github.com/openziti/ziti/issues/1726) - Fix some sdk hosting logging
* [Issue #1725](https://github.com/openziti/ziti/issues/1725) - Fix panic in entity event processing
# Release 0.32.1
## What's New
common/capabilities/consts.go
+1
View File
@@ -19,4 +19,5 @@ package capabilities
const (
ControllerCreateTerminatorV2 int = 1
ControllerSingleRouterLinkSource int = 2
ControllerCreateCircuitV2 int = 3
)
common/ctrl_msg/messages.go
+107
View File
@@ -17,7 +17,10 @@
package ctrl_msg
import (
"errors"
"fmt"
"github.com/openziti/channel/v2"
"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
)
const (
@@ -42,6 +45,16 @@ const (
ErrorTypeMisconfiguredTerminator = 2
ErrorTypeDialTimedOut = 3
ErrorTypeConnectionRefused = 4
CreateCircuitPeerDataHeader = 10
CreateCircuitReqSessionTokenHeader = 11
CreateCircuitReqFingerprintsHeader = 12
CreateCircuitReqTerminatorInstanceIdHeader = 13
CreateCircuitRespCircuitId = 11
CreateCircuitRespAddress = 12
CreateCircuitRespTagsHeader = 13
)
func NewCircuitSuccessMsg(sessionId, address string) *channel.Message {
@@ -68,3 +81,97 @@ func NewRouteResultFailedMessage(sessionId string, attempt int, rerr string) *ch
msg.Headers[RouteResultErrorHeader] = []byte(rerr)
return msg
}
type CreateCircuitRequest struct {
SessionToken string
Fingerprints []string
TerminatorInstanceId string
PeerData map[uint32][]byte
}
func (self *CreateCircuitRequest) GetSessionToken() string {
return self.SessionToken
}
func (self *CreateCircuitRequest) GetFingerprints() []string {
return self.Fingerprints
}
func (self *CreateCircuitRequest) GetTerminatorInstanceId() string {
return self.TerminatorInstanceId
}
func (self *CreateCircuitRequest) GetPeerData() map[uint32][]byte {
return self.PeerData
}
func (self *CreateCircuitRequest) ToMessage() *channel.Message {
msg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_CreateCircuitV2RequestType), nil)
msg.PutStringHeader(CreateCircuitReqSessionTokenHeader, self.SessionToken)
msg.PutStringSliceHeader(CreateCircuitReqFingerprintsHeader, self.Fingerprints)
msg.PutStringHeader(CreateCircuitReqTerminatorInstanceIdHeader, self.TerminatorInstanceId)
msg.PutU32ToBytesMapHeader(CreateCircuitPeerDataHeader, self.PeerData)
return msg
}
func DecodeCreateCircuitRequest(m *channel.Message) (*CreateCircuitRequest, error) {
sessionToken, _ := m.GetStringHeader(CreateCircuitReqSessionTokenHeader)
if len(sessionToken) == 0 {
return nil, errors.New("no session token provided in create circuit request")
}
fingerprints, _, err := m.GetStringSliceHeader(CreateCircuitReqFingerprintsHeader)
if err != nil {
return nil, fmt.Errorf("unable to get create circuit request fingerprints (%w)", err)
}
terminatorInstanceId, _ := m.GetStringHeader(CreateCircuitReqTerminatorInstanceIdHeader)
peerData, _, err := m.GetU32ToBytesMapHeader(CreateCircuitPeerDataHeader)
if err != nil {
return nil, fmt.Errorf("unable to get create circuit request peer data (%w)", err)
}
return &CreateCircuitRequest{
SessionToken: sessionToken,
Fingerprints: fingerprints,
TerminatorInstanceId: terminatorInstanceId,
PeerData: peerData,
}, nil
}
type CreateCircuitResponse struct {
CircuitId string
Address string
PeerData map[uint32][]byte
Tags map[string]string
}
func (self *CreateCircuitResponse) ToMessage() *channel.Message {
msg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_CreateCircuitV2ResponseType), nil)
msg.PutStringHeader(CreateCircuitRespCircuitId, self.CircuitId)
msg.PutStringHeader(CreateCircuitRespAddress, self.Address)
msg.PutU32ToBytesMapHeader(CreateCircuitPeerDataHeader, self.PeerData)
msg.PutStringToStringMapHeader(CreateCircuitRespTagsHeader, self.Tags)
return msg
}
func DecodeCreateCircuitResponse(m *channel.Message) (*CreateCircuitResponse, error) {
circuitId, _ := m.GetStringHeader(CreateCircuitRespCircuitId)
address, _ := m.GetStringHeader(CreateCircuitRespAddress)
peerData, _, err := m.GetU32ToBytesMapHeader(CreateCircuitPeerDataHeader)
if err != nil {
return nil, fmt.Errorf("unable to get create circuit response peer data (%w)", err)
}
tags, _, err := m.GetStringToStringMapHeader(CreateCircuitRespTagsHeader)
if err != nil {
return nil, fmt.Errorf("unable to get create circuit response tags (%w)", err)
}
return &CreateCircuitResponse{
CircuitId: circuitId,
Address: address,
PeerData: peerData,
Tags: tags,
}, nil
}
common/pb/edge_cmd_pb/edge_cmd.pb.go
+1 -1
View File
@@ -1,7 +1,7 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.31.0
// protoc v4.23.4
// protoc v3.21.12
// source: edge_cmd.proto
package edge_cmd_pb
common/pb/edge_ctrl_pb/edge_ctrl.pb.go
+77 -63
View File
@@ -48,6 +48,8 @@ const (
ContentType_SigningCertAddedType ContentType = 20215
ContentType_CreateTerminatorV2RequestType ContentType = 20216
ContentType_CreateTerminatorV2ResponseType ContentType = 20217
ContentType_CreateCircuitV2RequestType ContentType = 20218
ContentType_CreateCircuitV2ResponseType ContentType = 20219
ContentType_EnrollType ContentType = 20300
ContentType_EnrollmentCertsResponseType ContentType = 20301
ContentType_EnrollmentExtendRouterRequestType ContentType = 20302
@@ -93,6 +95,8 @@ var (
20215: "SigningCertAddedType",
20216: "CreateTerminatorV2RequestType",
20217: "CreateTerminatorV2ResponseType",
20218: "CreateCircuitV2RequestType",
20219: "CreateCircuitV2ResponseType",
20300: "EnrollType",
20301: "EnrollmentCertsResponseType",
20302: "EnrollmentExtendRouterRequestType",
@@ -135,6 +139,8 @@ var (
"SigningCertAddedType": 20215,
"CreateTerminatorV2RequestType": 20216,
"CreateTerminatorV2ResponseType": 20217,
"CreateCircuitV2RequestType": 20218,
"CreateCircuitV2ResponseType": 20219,
"EnrollType": 20300,
"EnrollmentCertsResponseType": 20301,
"EnrollmentExtendRouterRequestType": 20302,
@@ -381,6 +387,7 @@ const (
CreateTerminatorResult_Success CreateTerminatorResult = 0
CreateTerminatorResult_FailedIdConflict CreateTerminatorResult = 1
CreateTerminatorResult_FailedOther CreateTerminatorResult = 2
CreateTerminatorResult_FailedBusy CreateTerminatorResult = 3
)
// Enum value maps for CreateTerminatorResult.
@@ -389,11 +396,13 @@ var (
0: "Success",
1: "FailedIdConflict",
2: "FailedOther",
3: "FailedBusy",
}
CreateTerminatorResult_value = map[string]int32{
"Success": 0,
"FailedIdConflict": 1,
"FailedOther": 2,
"FailedBusy": 3,
}
)
@@ -3985,7 +3994,7 @@ var file_edge_ctrl_proto_rawDesc = []byte{
0x69, 0x66, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x24, 0x0a, 0x0d, 0x63, 0x6c,
0x69, 0x65, 0x6e, 0x74, 0x43, 0x65, 0x72, 0x74, 0x50, 0x65, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28,
0x09, 0x52, 0x0d, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x43, 0x65, 0x72, 0x74, 0x50, 0x65, 0x6d,
0x2a, 0xaf, 0x0a, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65,
0x2a, 0xf4, 0x0a, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65,
0x12, 0x08, 0x0a, 0x04, 0x5a, 0x65, 0x72, 0x6f, 0x10, 0x00, 0x12, 0x15, 0x0a, 0x0f, 0x53, 0x65,
0x72, 0x76, 0x65, 0x72, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x54, 0x79, 0x70, 0x65, 0x10, 0xa0, 0x9c,
0x01, 0x12, 0x15, 0x0a, 0x0f, 0x43, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x48, 0x65, 0x6c, 0x6c, 0x6f,
@@ -4028,68 +4037,73 @@ var file_edge_ctrl_proto_rawDesc = []byte{
0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x56, 0x32, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54,
0x79, 0x70, 0x65, 0x10, 0xf8, 0x9d, 0x01, 0x12, 0x24, 0x0a, 0x1e, 0x43, 0x72, 0x65, 0x61, 0x74,
0x65, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x56, 0x32, 0x52, 0x65, 0x73,
0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf9, 0x9d, 0x01, 0x12, 0x10, 0x0a,
0x0a, 0x45, 0x6e, 0x72, 0x6f, 0x6c, 0x6c, 0x54, 0x79, 0x70, 0x65, 0x10, 0xcc, 0x9e, 0x01, 0x12,
0x21, 0x0a, 0x1b, 0x45, 0x6e, 0x72, 0x6f, 0x6c, 0x6c, 0x6d, 0x65, 0x6e, 0x74, 0x43, 0x65, 0x72,
0x74, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xcd,
0x9e, 0x01, 0x12, 0x27, 0x0a, 0x21, 0x45, 0x6e, 0x72, 0x6f, 0x6c, 0x6c, 0x6d, 0x65, 0x6e, 0x74,
0x45, 0x78, 0x74, 0x65, 0x6e, 0x64, 0x52, 0x6f, 0x75, 0x74, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75,
0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xce, 0x9e, 0x01, 0x12, 0x2d, 0x0a, 0x27, 0x45,
0x6e, 0x72, 0x6f, 0x6c, 0x6c, 0x6d, 0x65, 0x6e, 0x74, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x64, 0x52,
0x6f, 0x75, 0x74, 0x65, 0x72, 0x56, 0x65, 0x72, 0x69, 0x66, 0x79, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xcf, 0x9e, 0x01, 0x12, 0x21, 0x0a, 0x1b, 0x43, 0x72,
0x65, 0x61, 0x74, 0x65, 0x41, 0x70, 0x69, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb0, 0x9f, 0x01, 0x12, 0x22, 0x0a,
0x1c, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x41, 0x70, 0x69, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f,
0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb1, 0x9f,
0x01, 0x12, 0x28, 0x0a, 0x22, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x43, 0x69, 0x72, 0x63, 0x75,
0x69, 0x74, 0x46, 0x6f, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75,
0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb2, 0x9f, 0x01, 0x12, 0x29, 0x0a, 0x23, 0x43,
0x72, 0x65, 0x61, 0x74, 0x65, 0x43, 0x69, 0x72, 0x63, 0x75, 0x69, 0x74, 0x46, 0x6f, 0x72, 0x53,
0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79,
0x70, 0x65, 0x10, 0xb3, 0x9f, 0x01, 0x12, 0x1d, 0x0a, 0x17, 0x4c, 0x69, 0x73, 0x74, 0x53, 0x65,
0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70,
0x65, 0x10, 0xb4, 0x9f, 0x01, 0x12, 0x15, 0x0a, 0x0f, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65,
0x4c, 0x69, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb5, 0x9f, 0x01, 0x12, 0x27, 0x0a, 0x21,
0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x54, 0x65, 0x72, 0x6d,
0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70,
0x65, 0x10, 0xb6, 0x9f, 0x01, 0x12, 0x28, 0x0a, 0x22, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54,
0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52,
0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb7, 0x9f, 0x01, 0x12,
0x27, 0x0a, 0x21, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x54,
0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
0x54, 0x79, 0x70, 0x65, 0x10, 0xb8, 0x9f, 0x01, 0x12, 0x28, 0x0a, 0x22, 0x55, 0x70, 0x64, 0x61,
0x74, 0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74,
0x6f, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb9,
0x9f, 0x01, 0x12, 0x27, 0x0a, 0x21, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x54, 0x75, 0x6e, 0x6e,
0x65, 0x6c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x75,
0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xba, 0x9f, 0x01, 0x12, 0x28, 0x0a, 0x22, 0x52,
0x65, 0x6d, 0x6f, 0x76, 0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x54, 0x65, 0x72, 0x6d, 0x69,
0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70,
0x65, 0x10, 0xbb, 0x9f, 0x01, 0x12, 0x1b, 0x0a, 0x15, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x48,
0x65, 0x61, 0x6c, 0x74, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xbc,
0x9f, 0x01, 0x2a, 0x21, 0x0a, 0x0b, 0x53, 0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x54, 0x79, 0x70,
0x65, 0x12, 0x08, 0x0a, 0x04, 0x44, 0x69, 0x61, 0x6c, 0x10, 0x00, 0x12, 0x08, 0x0a, 0x04, 0x42,
0x69, 0x6e, 0x64, 0x10, 0x01, 0x2a, 0x3d, 0x0a, 0x06, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x12,
0x0e, 0x0a, 0x0a, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x5a, 0x65, 0x72, 0x6f, 0x10, 0x00, 0x12,
0x11, 0x0a, 0x0c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x49, 0x64, 0x10,
0xfe, 0x07, 0x12, 0x10, 0x0a, 0x0b, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x50, 0x61, 0x73, 0x73, 0x65,
0x64, 0x10, 0xff, 0x07, 0x2a, 0x2e, 0x0a, 0x0a, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x54, 0x79,
0x70, 0x65, 0x12, 0x07, 0x0a, 0x03, 0x41, 0x64, 0x64, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x52,
0x65, 0x6d, 0x6f, 0x76, 0x65, 0x10, 0x01, 0x12, 0x0b, 0x0a, 0x07, 0x52, 0x65, 0x70, 0x6c, 0x61,
0x63, 0x65, 0x10, 0x02, 0x2a, 0x3d, 0x0a, 0x14, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74,
0x6f, 0x72, 0x50, 0x72, 0x65, 0x63, 0x65, 0x64, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x0b, 0x0a, 0x07,
0x44, 0x65, 0x66, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x00, 0x12, 0x0c, 0x0a, 0x08, 0x52, 0x65, 0x71,
0x75, 0x69, 0x72, 0x65, 0x64, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x46, 0x61, 0x69, 0x6c, 0x65,
0x64, 0x10, 0x02, 0x2a, 0x4c, 0x0a, 0x16, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x65, 0x72,
0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x0b, 0x0a,
0x07, 0x53, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x10, 0x00, 0x12, 0x14, 0x0a, 0x10, 0x46, 0x61,
0x69, 0x6c, 0x65, 0x64, 0x49, 0x64, 0x43, 0x6f, 0x6e, 0x66, 0x6c, 0x69, 0x63, 0x74, 0x10, 0x01,
0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x4f, 0x74, 0x68, 0x65, 0x72, 0x10,
0x02, 0x42, 0x2a, 0x5a, 0x28, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f,
0x6f, 0x70, 0x65, 0x6e, 0x7a, 0x69, 0x74, 0x69, 0x2f, 0x65, 0x64, 0x67, 0x65, 0x2f, 0x70, 0x62,
0x2f, 0x65, 0x64, 0x67, 0x65, 0x5f, 0x63, 0x74, 0x72, 0x6c, 0x5f, 0x70, 0x62, 0x62, 0x06, 0x70,
0x72, 0x6f, 0x74, 0x6f, 0x33,
0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xf9, 0x9d, 0x01, 0x12, 0x20, 0x0a,
0x1a, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x43, 0x69, 0x72, 0x63, 0x75, 0x69, 0x74, 0x56, 0x32,
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xfa, 0x9d, 0x01, 0x12,
0x21, 0x0a, 0x1b, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x43, 0x69, 0x72, 0x63, 0x75, 0x69, 0x74,
0x56, 0x32, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xfb,
0x9d, 0x01, 0x12, 0x10, 0x0a, 0x0a, 0x45, 0x6e, 0x72, 0x6f, 0x6c, 0x6c, 0x54, 0x79, 0x70, 0x65,
0x10, 0xcc, 0x9e, 0x01, 0x12, 0x21, 0x0a, 0x1b, 0x45, 0x6e, 0x72, 0x6f, 0x6c, 0x6c, 0x6d, 0x65,
0x6e, 0x74, 0x43, 0x65, 0x72, 0x74, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54,
0x79, 0x70, 0x65, 0x10, 0xcd, 0x9e, 0x01, 0x12, 0x27, 0x0a, 0x21, 0x45, 0x6e, 0x72, 0x6f, 0x6c,
0x6c, 0x6d, 0x65, 0x6e, 0x74, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x64, 0x52, 0x6f, 0x75, 0x74, 0x65,
0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xce, 0x9e, 0x01,
0x12, 0x2d, 0x0a, 0x27, 0x45, 0x6e, 0x72, 0x6f, 0x6c, 0x6c, 0x6d, 0x65, 0x6e, 0x74, 0x45, 0x78,
0x74, 0x65, 0x6e, 0x64, 0x52, 0x6f, 0x75, 0x74, 0x65, 0x72, 0x56, 0x65, 0x72, 0x69, 0x66, 0x79,
0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xcf, 0x9e, 0x01, 0x12,
0x21, 0x0a, 0x1b, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x41, 0x70, 0x69, 0x53, 0x65, 0x73, 0x73,
0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb0,
0x9f, 0x01, 0x12, 0x22, 0x0a, 0x1c, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x41, 0x70, 0x69, 0x53,
0x65, 0x73, 0x73, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79,
0x70, 0x65, 0x10, 0xb1, 0x9f, 0x01, 0x12, 0x28, 0x0a, 0x22, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65,
0x43, 0x69, 0x72, 0x63, 0x75, 0x69, 0x74, 0x46, 0x6f, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63,
0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb2, 0x9f, 0x01,
0x12, 0x29, 0x0a, 0x23, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x43, 0x69, 0x72, 0x63, 0x75, 0x69,
0x74, 0x46, 0x6f, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f,
0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb3, 0x9f, 0x01, 0x12, 0x1d, 0x0a, 0x17, 0x4c,
0x69, 0x73, 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb4, 0x9f, 0x01, 0x12, 0x15, 0x0a, 0x0f, 0x53, 0x65,
0x72, 0x76, 0x69, 0x63, 0x65, 0x4c, 0x69, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb5, 0x9f,
0x01, 0x12, 0x27, 0x0a, 0x21, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65,
0x6c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb6, 0x9f, 0x01, 0x12, 0x28, 0x0a, 0x22, 0x43, 0x72,
0x65, 0x61, 0x74, 0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e,
0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54, 0x79, 0x70, 0x65,
0x10, 0xb7, 0x9f, 0x01, 0x12, 0x27, 0x0a, 0x21, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x75,
0x6e, 0x6e, 0x65, 0x6c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65,
0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xb8, 0x9f, 0x01, 0x12, 0x28, 0x0a,
0x22, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x54, 0x65, 0x72,
0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x54,
0x79, 0x70, 0x65, 0x10, 0xb9, 0x9f, 0x01, 0x12, 0x27, 0x0a, 0x21, 0x52, 0x65, 0x6d, 0x6f, 0x76,
0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f,
0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x54, 0x79, 0x70, 0x65, 0x10, 0xba, 0x9f, 0x01,
0x12, 0x28, 0x0a, 0x22, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x54, 0x75, 0x6e, 0x6e, 0x65, 0x6c,
0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
0x73, 0x65, 0x54, 0x79, 0x70, 0x65, 0x10, 0xbb, 0x9f, 0x01, 0x12, 0x1b, 0x0a, 0x15, 0x54, 0x75,
0x6e, 0x6e, 0x65, 0x6c, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x54,
0x79, 0x70, 0x65, 0x10, 0xbc, 0x9f, 0x01, 0x2a, 0x21, 0x0a, 0x0b, 0x53, 0x65, 0x73, 0x73, 0x69,
0x6f, 0x6e, 0x54, 0x79, 0x70, 0x65, 0x12, 0x08, 0x0a, 0x04, 0x44, 0x69, 0x61, 0x6c, 0x10, 0x00,
0x12, 0x08, 0x0a, 0x04, 0x42, 0x69, 0x6e, 0x64, 0x10, 0x01, 0x2a, 0x3d, 0x0a, 0x06, 0x48, 0x65,
0x61, 0x64, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x0a, 0x48, 0x65, 0x61, 0x64, 0x65, 0x72, 0x5a, 0x65,
0x72, 0x6f, 0x10, 0x00, 0x12, 0x11, 0x0a, 0x0c, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74,
0x6f, 0x72, 0x49, 0x64, 0x10, 0xfe, 0x07, 0x12, 0x10, 0x0a, 0x0b, 0x43, 0x68, 0x65, 0x63, 0x6b,
0x50, 0x61, 0x73, 0x73, 0x65, 0x64, 0x10, 0xff, 0x07, 0x2a, 0x2e, 0x0a, 0x0a, 0x43, 0x6f, 0x6e,
0x66, 0x69, 0x67, 0x54, 0x79, 0x70, 0x65, 0x12, 0x07, 0x0a, 0x03, 0x41, 0x64, 0x64, 0x10, 0x00,
0x12, 0x0a, 0x0a, 0x06, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x10, 0x01, 0x12, 0x0b, 0x0a, 0x07,
0x52, 0x65, 0x70, 0x6c, 0x61, 0x63, 0x65, 0x10, 0x02, 0x2a, 0x3d, 0x0a, 0x14, 0x54, 0x65, 0x72,
0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x50, 0x72, 0x65, 0x63, 0x65, 0x64, 0x65, 0x6e, 0x63,
0x65, 0x12, 0x0b, 0x0a, 0x07, 0x44, 0x65, 0x66, 0x61, 0x75, 0x6c, 0x74, 0x10, 0x00, 0x12, 0x0c,
0x0a, 0x08, 0x52, 0x65, 0x71, 0x75, 0x69, 0x72, 0x65, 0x64, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06,
0x46, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x10, 0x02, 0x2a, 0x5c, 0x0a, 0x16, 0x43, 0x72, 0x65, 0x61,
0x74, 0x65, 0x54, 0x65, 0x72, 0x6d, 0x69, 0x6e, 0x61, 0x74, 0x6f, 0x72, 0x52, 0x65, 0x73, 0x75,
0x6c, 0x74, 0x12, 0x0b, 0x0a, 0x07, 0x53, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, 0x10, 0x00, 0x12,
0x14, 0x0a, 0x10, 0x46, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x49, 0x64, 0x43, 0x6f, 0x6e, 0x66, 0x6c,
0x69, 0x63, 0x74, 0x10, 0x01, 0x12, 0x0f, 0x0a, 0x0b, 0x46, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x4f,
0x74, 0x68, 0x65, 0x72, 0x10, 0x02, 0x12, 0x0e, 0x0a, 0x0a, 0x46, 0x61, 0x69, 0x6c, 0x65, 0x64,
0x42, 0x75, 0x73, 0x79, 0x10, 0x03, 0x42, 0x2a, 0x5a, 0x28, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62,
0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6f, 0x70, 0x65, 0x6e, 0x7a, 0x69, 0x74, 0x69, 0x2f, 0x65, 0x64,
0x67, 0x65, 0x2f, 0x70, 0x62, 0x2f, 0x65, 0x64, 0x67, 0x65, 0x5f, 0x63, 0x74, 0x72, 0x6c, 0x5f,
0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
common/pb/edge_ctrl_pb/edge_ctrl.proto
+3
View File
@@ -30,6 +30,8 @@ enum ContentType {
SigningCertAddedType = 20215;
CreateTerminatorV2RequestType = 20216;
CreateTerminatorV2ResponseType = 20217;
CreateCircuitV2RequestType = 20218;
CreateCircuitV2ResponseType = 20219;
EnrollType = 20300;
EnrollmentCertsResponseType = 20301;
@@ -241,6 +243,7 @@ enum CreateTerminatorResult {
Success = 0;
FailedIdConflict = 1;
FailedOther = 2;
FailedBusy = 3;
}
message CreateTerminatorV2Response {
common/pb/edge_ctrl_pb/messages.go
+1 -1
View File
@@ -2,8 +2,8 @@ package edge_ctrl_pb
import (
"github.com/openziti/edge-api/rest_model"
"github.com/openziti/ziti/controller/xt"
"github.com/openziti/sdk-golang/ziti"
"github.com/openziti/ziti/controller/xt"
)
func (m *ClientHello) GetContentType() int32 {
controller/command/rate_limiter.go
+9
View File
@@ -18,6 +18,7 @@ package command
import (
"fmt"
"github.com/openziti/foundation/v2/errorz"
"github.com/openziti/metrics"
"github.com/openziti/ziti/controller/apierror"
"github.com/pkg/errors"
@@ -372,3 +373,11 @@ type noOpRateLimitControl struct{}
func (noOpRateLimitControl) Success() {}
func (noOpRateLimitControl) Timeout() {}
func WasRateLimited(err error) bool {
var apiErr *errorz.ApiError
if errors.As(err, &apiErr) {
return apiErr.Code == apierror.ServerTooManyRequestsCode
}
return false
}
controller/command/rate_limiter_test.go
+9
View File
@@ -25,6 +25,7 @@ import (
"github.com/openziti/metrics"
"github.com/openziti/sdk-golang/ziti"
"github.com/openziti/ziti/controller/apierror"
"github.com/stretchr/testify/assert"
"sync"
"sync/atomic"
"testing"
@@ -146,3 +147,11 @@ func Test_AuthFlood(t *testing.T) {
countdown.Wait()
}
func Test_WasApiRateLimited(t *testing.T) {
err := errors.New("hello")
assert.False(t, WasRateLimited(err))
err = apierror.NewTooManyUpdatesError()
assert.True(t, WasRateLimited(err))
}
controller/config.go
+1
View File
@@ -462,6 +462,7 @@ func LoadConfig(path string) (*Config, error) {
}
}
controllerConfig.CommandRateLimiter.Enabled = true
controllerConfig.CommandRateLimiter.QueueSize = command.DefaultLimiterSize
if value, found := cfgmap["commandRateLimiter"]; found {
controller/controller.go
+1
View File
@@ -265,6 +265,7 @@ func (c *Controller) Run() error {
capabilityMask := &big.Int{}
capabilityMask.SetBit(capabilityMask, capabilities.ControllerCreateTerminatorV2, 1)
capabilityMask.SetBit(capabilityMask, capabilities.ControllerSingleRouterLinkSource, 1)
capabilityMask.SetBit(capabilityMask, capabilities.ControllerCreateCircuitV2, 1)
headers := map[int32][]byte{
channel.HelloVersionHeader: versionHeader,
int32(ctrl_pb.ControlHeaders_CapabilitiesHeader): capabilityMask.Bytes(),
controller/events/dispatcher_entity_change.go
+1 -1
View File
@@ -354,7 +354,7 @@ func (self *entityChangeEventFilter) AcceptEntityChangeEvent(evt *event.EntityCh
return
}
if *evt.IsParentEvent && !self.includeParentEvents {
if evt.IsParentEvent != nil && *evt.IsParentEvent && !self.includeParentEvents {
return
}
}
controller/handler_edge_ctrl/common.go
-1
View File
@@ -27,7 +27,6 @@ type requestHandler interface {
getAppEnv() *env.AppEnv
getNetwork() *network.Network
getChannel() channel.Channel
ContentType() int32
Label() string
}
controller/handler_edge_ctrl/create_circuit.go
+95 -42
View File
@@ -17,11 +17,14 @@
package handler_edge_ctrl
import (
"fmt"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel/v2"
"github.com/openziti/ziti/common/ctrl_msg"
"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
"github.com/openziti/ziti/controller/db"
"github.com/openziti/ziti/controller/env"
"github.com/openziti/ziti/controller/network"
"google.golang.org/protobuf/proto"
)
@@ -30,39 +33,36 @@ type createCircuitHandler struct {
}
func NewCreateCircuitHandler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler {
return &createCircuitHandler{
handler := &createCircuitHandler{
baseRequestHandler: baseRequestHandler{
ch: ch,
appEnv: appEnv,
},
}
return &channel.AsyncFunctionReceiveAdapter{
Type: int32(edge_ctrl_pb.ContentType_CreateCircuitRequestType),
Handler: handler.HandleReceiveCreateCircuitV1,
}
}
func (self *createCircuitHandler) ContentType() int32 {
return int32(edge_ctrl_pb.ContentType_CreateCircuitRequestType)
func NewCreateCircuitV2Handler(appEnv *env.AppEnv, ch channel.Channel) channel.TypedReceiveHandler {
handler := &createCircuitHandler{
baseRequestHandler: baseRequestHandler{
ch: ch,
appEnv: appEnv,
},
}
return &channel.AsyncFunctionReceiveAdapter{
Type: int32(edge_ctrl_pb.ContentType_CreateCircuitV2RequestType),
Handler: handler.HandleReceiveCreateCircuitV2,
}
}
func (self *createCircuitHandler) Label() string {
return "create.circuit"
}
func (self *createCircuitHandler) sendResponse(ctx *CreateCircuitRequestContext, response *edge_ctrl_pb.CreateCircuitResponse) {
log := pfxlog.ContextLogger(self.ch.Label())
body, err := proto.Marshal(response)
if err != nil {
log.WithError(err).WithField("token", ctx.req.SessionToken).Error("failed to marshal create circuit response")
return
}
responseMsg := channel.NewMessage(response.GetContentType(), body)
responseMsg.ReplyTo(ctx.msg)
if err = self.ch.Send(responseMsg); err != nil {
log.WithError(err).WithField("token", ctx.req.SessionToken).Error("failed to send create circuit response")
}
}
func (self *createCircuitHandler) HandleReceive(msg *channel.Message, ch channel.Channel) {
func (self *createCircuitHandler) HandleReceiveCreateCircuitV1(msg *channel.Message, ch channel.Channel) {
req := &edge_ctrl_pb.CreateCircuitRequest{}
if err := proto.Unmarshal(msg.Body, req); err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not unmarshal CreateCircuitRequest")
@@ -74,27 +74,10 @@ func (self *createCircuitHandler) HandleReceive(msg *channel.Message, ch channel
req: req,
}
go self.CreateCircuit(ctx)
self.CreateCircuit(ctx, self.CreateCircuitV1Response)
}
func (self *createCircuitHandler) CreateCircuit(ctx *CreateCircuitRequestContext) {
if !ctx.loadRouter() {
return
}
ctx.loadSession(ctx.req.SessionToken)
ctx.checkSessionType(db.SessionTypeDial)
ctx.checkSessionFingerprints(ctx.req.Fingerprints)
ctx.verifyEdgeRouterAccess()
ctx.loadService()
circuitInfo, peerData := ctx.createCircuit(ctx.req.TerminatorInstanceId, ctx.req.PeerData)
if ctx.err != nil {
self.returnError(ctx, ctx.err)
return
}
log := pfxlog.ContextLogger(self.ch.Label()).WithField("token", ctx.req.SessionToken)
func (self *createCircuitHandler) CreateCircuitV1Response(circuitInfo *network.Circuit, peerData map[uint32][]byte) (*channel.Message, error) {
response := &edge_ctrl_pb.CreateCircuitResponse{
CircuitId: circuitInfo.Id,
Address: circuitInfo.Path.IngressId,
@@ -102,15 +85,85 @@ func (self *createCircuitHandler) CreateCircuit(ctx *CreateCircuitRequestContext
Tags: circuitInfo.Tags,
}
body, err := proto.Marshal(response)
if err != nil {
return nil, fmt.Errorf("failed to marshal protobuf create circuit response (%w)", err)
}
responseMsg := channel.NewMessage(response.GetContentType(), body)
return responseMsg, nil
}
func (self *createCircuitHandler) HandleReceiveCreateCircuitV2(msg *channel.Message, ch channel.Channel) {
req, err := ctrl_msg.DecodeCreateCircuitRequest(msg)
if err != nil {
pfxlog.ContextLogger(ch.Label()).WithError(err).Error("could not decode CreateCircuitRequest")
return
}
ctx := &CreateCircuitRequestContext{
baseSessionRequestContext: baseSessionRequestContext{handler: self, msg: msg},
req: req,
}
self.CreateCircuit(ctx, self.CreateCircuitV2Response)
}
func (self *createCircuitHandler) CreateCircuitV2Response(circuitInfo *network.Circuit, peerData map[uint32][]byte) (*channel.Message, error) {
response := &ctrl_msg.CreateCircuitResponse{
CircuitId: circuitInfo.Id,
Address: circuitInfo.Path.IngressId,
PeerData: peerData,
Tags: circuitInfo.Tags,
}
return response.ToMessage(), nil
}
func (self *createCircuitHandler) CreateCircuit(ctx *CreateCircuitRequestContext, f createCircuitResponseFactory) {
if !ctx.loadRouter() {
return
}
ctx.loadSession(ctx.req.GetSessionToken())
ctx.checkSessionType(db.SessionTypeDial)
ctx.checkSessionFingerprints(ctx.req.GetFingerprints())
ctx.verifyEdgeRouterAccess()
ctx.loadService()
circuitInfo, peerData := ctx.createCircuit(ctx.req.GetTerminatorInstanceId(), ctx.req.GetPeerData())
if ctx.err != nil {
self.returnError(ctx, ctx.err)
return
}
log := pfxlog.ContextLogger(self.ch.Label()).WithField("token", ctx.req.GetSessionToken())
responseMsg, err := f(circuitInfo, peerData)
if err != nil {
log.WithError(err).Error("error generating create circuit response")
}
log.Debugf("responding with successful circuit setup")
self.sendResponse(ctx, response)
responseMsg.ReplyTo(ctx.msg)
if err = self.ch.Send(responseMsg); err != nil {
log.WithError(err).WithField("token", ctx.req.GetSessionToken()).Error("failed to send create circuit response")
}
}
type createCircuitResponseFactory func(*network.Circuit, map[uint32][]byte) (*channel.Message, error)
var _ CreateCircuitRequest = (*edge_ctrl_pb.CreateCircuitRequest)(nil)
type CreateCircuitRequest interface {
GetSessionToken() string
GetFingerprints() []string
GetTerminatorInstanceId() string
GetPeerData() map[uint32][]byte
}
type CreateCircuitRequestContext struct {
baseSessionRequestContext
req *edge_ctrl_pb.CreateCircuitRequest
req CreateCircuitRequest
}
func (self *CreateCircuitRequestContext) GetSessionToken() string {
return self.req.SessionToken
return self.req.GetSessionToken()
}
controller/handler_edge_ctrl/create_terminator_v2.go
+5
View File
@@ -23,6 +23,7 @@ import (
"github.com/openziti/channel/v2/protobufs"
"github.com/openziti/ziti/common"
"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
"github.com/openziti/ziti/controller/command"
"github.com/openziti/ziti/controller/db"
"github.com/openziti/ziti/controller/env"
"github.com/openziti/ziti/controller/fields"
@@ -153,6 +154,10 @@ func (self *createTerminatorV2Handler) CreateTerminatorV2(ctx *CreateTerminatorV
return
}
} else {
if command.WasRateLimited(err) {
self.returnError(ctx, edge_ctrl_pb.CreateTerminatorResult_FailedBusy, err, logger)
return
}
self.returnError(ctx, edge_ctrl_pb.CreateTerminatorResult_FailedOther, err, logger)
return
}
controller/server/controller.go
+4 -3
View File
@@ -21,9 +21,9 @@ import (
"github.com/openziti/channel/v2"
"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
runner2 "github.com/openziti/ziti/common/runner"
"github.com/openziti/ziti/controller/api_impl"
"github.com/openziti/ziti/controller/handler_edge_mgmt"
sync2 "github.com/openziti/ziti/controller/sync_strats"
"github.com/openziti/ziti/controller/api_impl"
"os"
"sync"
"time"
@@ -31,13 +31,13 @@ import (
"github.com/openziti/ziti/controller/internal/policy"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/storage/boltz"
"github.com/openziti/ziti/common/config"
edgeconfig "github.com/openziti/ziti/controller/config"
"github.com/openziti/ziti/controller/env"
"github.com/openziti/ziti/controller/handler_edge_ctrl"
_ "github.com/openziti/ziti/controller/internal/routes"
"github.com/openziti/ziti/controller/model"
"github.com/openziti/ziti/common/config"
"github.com/openziti/storage/boltz"
)
type Controller struct {
@@ -153,6 +153,7 @@ func (c *Controller) GetCtrlHandlers(binding channel.Binding) []channel.TypedRec
result := []channel.TypedReceiveHandler{
handler_edge_ctrl.NewSessionHeartbeatHandler(c.AppEnv),
handler_edge_ctrl.NewCreateCircuitHandler(c.AppEnv, ch),
handler_edge_ctrl.NewCreateCircuitV2Handler(c.AppEnv, ch),
handler_edge_ctrl.NewCreateTerminatorHandler(c.AppEnv, ch),
handler_edge_ctrl.NewCreateTerminatorV2Handler(c.AppEnv, ch),
handler_edge_ctrl.NewUpdateTerminatorHandler(c.AppEnv, ch),
go.mod
+12 -12
View File
@@ -28,7 +28,7 @@ require (
github.com/golang-jwt/jwt/v5 v5.2.0
github.com/google/go-cmp v0.6.0
github.com/google/gopacket v1.1.19
github.com/google/uuid v1.5.0
github.com/google/uuid v1.6.0
github.com/gorilla/handlers v1.5.2
github.com/gorilla/mux v1.8.1
github.com/gorilla/websocket v1.5.1
@@ -47,17 +47,17 @@ require (
github.com/mitchellh/mapstructure v1.5.0
github.com/natefinch/lumberjack v2.0.0+incompatible
github.com/openziti/agent v1.0.16
github.com/openziti/channel/v2 v2.0.117
github.com/openziti/channel/v2 v2.0.119
github.com/openziti/edge-api v0.26.10
github.com/openziti/foundation/v2 v2.0.36
github.com/openziti/identity v1.0.69
github.com/openziti/foundation/v2 v2.0.37
github.com/openziti/identity v1.0.70
github.com/openziti/jwks v1.0.3
github.com/openziti/metrics v1.2.43
github.com/openziti/runzmd v1.0.37
github.com/openziti/sdk-golang v0.22.21
github.com/openziti/metrics v1.2.45
github.com/openziti/runzmd v1.0.38
github.com/openziti/sdk-golang v0.22.28
github.com/openziti/secretstream v0.1.16
github.com/openziti/storage v0.2.28
github.com/openziti/transport/v2 v2.0.121
github.com/openziti/storage v0.2.30
github.com/openziti/transport/v2 v2.0.122
github.com/openziti/x509-claims v1.0.3
github.com/openziti/xweb/v2 v2.1.0
github.com/openziti/ziti-db-explorer v1.1.3
@@ -66,7 +66,7 @@ require (
github.com/rabbitmq/amqp091-go v1.8.1
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
github.com/russross/blackfriday v1.6.0
github.com/shirou/gopsutil/v3 v3.23.12
github.com/shirou/gopsutil/v3 v3.24.1
github.com/sirupsen/logrus v1.9.3
github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e
github.com/spf13/cobra v1.8.0
@@ -81,7 +81,7 @@ require (
golang.org/x/crypto v0.18.0
golang.org/x/net v0.20.0
golang.org/x/sync v0.6.0
golang.org/x/sys v0.16.0
golang.org/x/sys v0.17.0
golang.org/x/text v0.14.0
google.golang.org/protobuf v1.32.0
gopkg.in/AlecAivazis/survey.v1 v1.8.7
@@ -184,7 +184,7 @@ require (
golang.org/x/image v0.13.0 // indirect
golang.org/x/mod v0.14.0 // indirect
golang.org/x/oauth2 v0.16.0 // indirect
golang.org/x/term v0.16.0 // indirect
golang.org/x/term v0.17.0 // indirect
golang.org/x/tools v0.16.0 // indirect
google.golang.org/appengine v1.6.8 // indirect
gopkg.in/ini.v1 v1.67.0 // indirect
go.sum
+24 -24
View File
@@ -324,8 +324,8 @@ github.com/google/pprof v0.0.0-20210122040257-d980be63207e/go.mod h1:kpwsk12EmLe
github.com/google/pprof v0.0.0-20210226084205-cbba55b83ad5/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
github.com/googleapis/gax-go/v2 v2.0.3/go.mod h1:LLvjysVCY1JZeum8Z6l8qUty8fiNwE08qbEPm1M08qg=
github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
@@ -565,30 +565,30 @@ github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYr
github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8=
github.com/openziti/agent v1.0.16 h1:9Saji+8hFE1NpzP2XzDhsVJbCrDlhixoLHfOpFt5Z+U=
github.com/openziti/agent v1.0.16/go.mod h1:zfm53+PVWoGFzjGGgQdKby5749G6VRYHe+eQJmoVKy4=
github.com/openziti/channel/v2 v2.0.117 h1:9EUNJIA4NCosn+Uci0gYNiDMsGBX52Jd34wpziW+zPA=
github.com/openziti/channel/v2 v2.0.117/go.mod h1:RADBBOkbwpFMmbznU3bZmUsBB646seqZnQ7GPI6Chjg=
github.com/openziti/channel/v2 v2.0.119 h1:stfSrnDqoTi78LMvQA3+NSivHjQnRrYKrgij5NaOENI=
github.com/openziti/channel/v2 v2.0.119/go.mod h1:lSRJwqmbkE34DgXYEmUhVCzwcQcx65vZGE8nuBNK458=
github.com/openziti/dilithium v0.3.3 h1:PLgQ6PMNLSTzCFbX/h98cmudgz/cU6TmjdSv5NAPD8k=
github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo=
github.com/openziti/edge-api v0.26.10 h1:LEDuJHZsExi0PBVO9iVuIdZWJ7eFo/i4TJhXoSFmfOU=
github.com/openziti/edge-api v0.26.10/go.mod h1:FQLjav9AfqxQYSL0xKPDZ/JWTSZXApkk7jM2/iczGXM=
github.com/openziti/foundation/v2 v2.0.36 h1:ogEIvsWur8/9mUzf9NOB4hRUyx372Uy6AmnHRcurIkY=
github.com/openziti/foundation/v2 v2.0.36/go.mod h1:MdK2oAJSwo7iCfvVdG16ZGz47qP7nG97ovnqEdXW2kQ=
github.com/openziti/identity v1.0.69 h1:wNgQomnv8ar2S1wge9jQK1jpqE2virOKKG8GyfTiHMQ=
github.com/openziti/identity v1.0.69/go.mod h1:+7hQNAG5nUUdz0165OubWpMHkP/ZVJB3nv9HUXXi9H8=
github.com/openziti/foundation/v2 v2.0.37 h1:7pa4vWrlwllEoLXaK2rx91AffLQJ8k5pvc92oWANavA=
github.com/openziti/foundation/v2 v2.0.37/go.mod h1:2NxzCnJbMw35U9RrFcdEaiXdxIMfBHOUNPngpyhvKeY=
github.com/openziti/identity v1.0.70 h1:JNwtJHmIS0DcXookm2xuXyh4z92T1O21GQvuO8PmHWs=
github.com/openziti/identity v1.0.70/go.mod h1:jsKBL4G1BsmDSCIfhK4jha5B3Sevgy1jyZq0GtFKhSk=
github.com/openziti/jwks v1.0.3 h1:hf8wkb+Cg4nH/HM0KROFd7u+C3DkRVcFZJ7tDV+4icc=
github.com/openziti/jwks v1.0.3/go.mod h1:t4xxq8vlXGsPn29kiQVnZBBDDnEoOFqtJoHibkJunQQ=
github.com/openziti/metrics v1.2.43 h1:DSrmpLhoA45DlLVNdKOn2lBfCM0/r6wKz+3SDXe8X7Y=
github.com/openziti/metrics v1.2.43/go.mod h1:+RY4avT60Vbxb9wyfvRD0msrARyYCB5+heb8VIZzCm8=
github.com/openziti/runzmd v1.0.37 h1:qj2r9z4t7OAdmIXMdGbP9Su6TqA0bLdD2RMjJ71LRS0=
github.com/openziti/runzmd v1.0.37/go.mod h1:eKhqJsGoLeDHex/o5Mw6TcNJxlVljafSVm7ZU+bX5G8=
github.com/openziti/sdk-golang v0.22.21 h1:rj8gCq9SPcRpStPX0GUL8X2vkCQNUn+nk2WbyY82b/Y=
github.com/openziti/sdk-golang v0.22.21/go.mod h1:w3M9yAVnW7IAU1dpYtiFB/mIYLiU7l4tB+c4coqWRwo=
github.com/openziti/metrics v1.2.45 h1:+3zqszLWyFdTgzbsQD90V0yJcC9Ek77qKaIGMQXkAXs=
github.com/openziti/metrics v1.2.45/go.mod h1:g6CgAEbFes2UtdfGrsR8AKkuoBVL5dkU61843uQvllM=
github.com/openziti/runzmd v1.0.38 h1:0eFWAf7R9Thx99ue7Gql2dBavsqv4jldU8W5AIZJ8wo=
github.com/openziti/runzmd v1.0.38/go.mod h1:WiQi+YIXxZBH1bwjr+eo/e3ftfTLEABpN0i2QIhBI9w=
github.com/openziti/sdk-golang v0.22.28 h1:s159CT42dXug4GiJiN/kM6/ol+N2LFZ2tUk6bOpbgiI=
github.com/openziti/sdk-golang v0.22.28/go.mod h1:BLaLvcLqAgf3JFoDPWLTj3j3X5rndo6ZejdDdkMlihQ=
github.com/openziti/secretstream v0.1.16 h1:tVanF7OpJL1MJ1gvWaRlR2i+kAbrGsxr3q6EXFOS08U=
github.com/openziti/secretstream v0.1.16/go.mod h1:bvjGBUW/0e5MzD5S3FW3rhGASRNWAi+kTkTENZ9qRDE=
github.com/openziti/storage v0.2.28 h1:qHnsSF4RgQpT23hOXlwkAvJ0gO5PLmvTqyBy13dm3Rc=
github.com/openziti/storage v0.2.28/go.mod h1:ahdvsmmdQWFsZXeExvXeA8MHUXPLrwuDFBJDTztLY5E=
github.com/openziti/transport/v2 v2.0.121 h1:9WB1/9YMFed2UuefqZq+odUYYkoFOr+8aBIAZEC1MNE=
github.com/openziti/transport/v2 v2.0.121/go.mod h1:hS7TzoyTIwm4lVEsGyjQkoEg+vA+6dAOR8S4FZmWIj4=
github.com/openziti/storage v0.2.30 h1:15o8rSSgtcNCSBONt81ZRASQXJHh2L9Y0aWcHoQ81dE=
github.com/openziti/storage v0.2.30/go.mod h1:ltsgk2G0c3jVZVTK5bHkRtblGdFFoSJ3fOMur/0iFKg=
github.com/openziti/transport/v2 v2.0.122 h1:XWwZ6JcSO1nvbZgfp6kdf8aR5LEEN343mpZlhSihirk=
github.com/openziti/transport/v2 v2.0.122/go.mod h1:07Ak2jMsyZmi7/ECxGNfMXk8cF1Vj5qtKj90FoAnK8A=
github.com/openziti/x509-claims v1.0.3 h1:HNdQ8Nf1agB3lBs1gahcO6zfkeS4S5xoQ2/PkY4HRX0=
github.com/openziti/x509-claims v1.0.3/go.mod h1:Z0WIpBm6c4ecrpRKrou6Gk2wrLWxJO/+tuUwKh8VewE=
github.com/openziti/xweb/v2 v2.1.0 h1:Xhh3C2pZkq/Prr65V+SfFSibLDYteoc4f62KQCcTZF4=
@@ -673,8 +673,8 @@ github.com/sagikazarmark/slog-shim v0.1.0 h1:diDBnUNK9N/354PgrxMywXnAwEr1QZcOr6g
github.com/sagikazarmark/slog-shim v0.1.0/go.mod h1:SrcSrq8aKtyuqEI1uvTDTK1arOWRIczQRv+GVI1AkeQ=
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
github.com/shirou/gopsutil/v3 v3.23.12 h1:z90NtUkp3bMtmICZKpC4+WaknU1eXtp5vtbQ11DgpE4=
github.com/shirou/gopsutil/v3 v3.23.12/go.mod h1:1FrWgea594Jp7qmjHUUPlJDTPgcsb9mGnXDxavtikzM=
github.com/shirou/gopsutil/v3 v3.24.1 h1:R3t6ondCEvmARp3wxODhXMTLC/klMa87h2PHUw5m7QI=
github.com/shirou/gopsutil/v3 v3.24.1/go.mod h1:UU7a2MSBQa+kW1uuDq8DeEBS8kmrnQwsv2b5O513rwU=
github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM=
github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
@@ -1078,17 +1078,17 @@ golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
golang.org/x/term v0.16.0 h1:m+B6fahuftsE9qjo0VWp2FW0mB3MTJvR0BaMQrq0pmE=
golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
golang.org/x/term v0.17.0 h1:mkTF7LCd6WGJNL3K1Ad7kwxNfYAW6a8a8QqtMblp/4U=
golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk=
golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
router/xgress/link_send_buffer.go
+3 -1
View File
@@ -99,7 +99,9 @@ func (self *txPayload) isRetransmittable() bool {
}
func NewLinkSendBuffer(x *Xgress) *LinkSendBuffer {
logrus.Debugf("txPortalStartSize = %d", x.Options.TxPortalStartSize)
logrus.Debugf("txPortalStartSize = %d, txPortalMinSize = %d",
x.Options.TxPortalStartSize,
x.Options.TxPortalMinSize)
// newlyBuffered should be size 0, otherwise payloads can be sent and acks received before the payload is
// processed by the LinkSendBuffer
router/xgress_edge/dialer.go
-7
View File
@@ -45,13 +45,6 @@ func (dialer *dialer) InspectTerminator(id string, destination string, fixInvali
pfxlog.Logger().Debug("looking up hosted service conn")
terminator, found := dialer.factory.hostedServices.Get(terminatorAddress)
if found && terminator.terminatorId.Load() == id {
updated := terminator.state.CompareAndSwap(TerminatorStateEstablishing, TerminatorStateEstablished) ||
terminator.state.CompareAndSwap(TerminatorStatePendingEstablishment, TerminatorStateEstablished)
if updated {
dialer.factory.hostedServices.notifyTerminatorCreated(id)
}
result, err := terminator.inspect(fixInvalid)
if err != nil {
return true, err.Error()
router/xgress_edge/fabric.go
+47 -19
View File
@@ -27,6 +27,7 @@ import (
"github.com/pkg/errors"
"io"
"math"
"math/rand"
"sync/atomic"
"time"
)
@@ -80,7 +81,30 @@ type edgeTerminator struct {
v2 bool
state concurrenz.AtomicValue[terminatorState]
postValidate bool
notifyEstablished bool
nextAttempt time.Time
retryDelay time.Duration
establishActive atomic.Bool
createTime time.Time
establishCallback func(ok bool, msg string)
}
func (self *edgeTerminator) calculateRetry(queueFailed bool) {
retryDelay := time.Second
if !queueFailed {
if self.retryDelay < time.Second {
self.retryDelay = time.Second
} else {
self.retryDelay = self.retryDelay * 2
if self.retryDelay > 30*time.Second {
self.retryDelay = 30 * time.Second
}
}
retryDelay = self.retryDelay
}
actualDelay := (float64(retryDelay.Milliseconds()) * 1.5) - float64(rand.Intn(int(retryDelay.Milliseconds())))
self.nextAttempt = time.Now().Add(time.Millisecond * time.Duration(actualDelay))
}
func (self *edgeTerminator) inspect(fixInvalidTerminators bool) (*edge.InspectResult, error) {
@@ -92,7 +116,7 @@ func (self *edgeTerminator) inspect(fixInvalidTerminators bool) (*edge.InspectRe
}
result, err := edge.UnmarshalInspectResult(resp)
if result != nil && result.Type != edge.ConnTypeBind && fixInvalidTerminators {
self.close(true, "terminator invalid")
self.close(true, true, "terminator invalid")
}
return result, err
}
@@ -106,13 +130,13 @@ func (self *edgeTerminator) nextDialConnId() uint32 {
return nextId
}
func (self *edgeTerminator) close(notify bool, reason string) {
func (self *edgeTerminator) close(notifySdk bool, notifyCtrl bool, reason string) {
logger := pfxlog.Logger().
WithField("terminatorId", self.terminatorId.Load()).
WithField("token", self.token).
WithField("reason", reason)
if notify && !self.IsClosed() {
if notifySdk && !self.IsClosed() {
// Notify edge client of close
logger.Debug("sending closed to SDK client")
closeMsg := edge.NewStateClosedMsg(self.Id(), reason)
@@ -129,6 +153,24 @@ func (self *edgeTerminator) close(notify bool, reason string) {
self.state.Store(TerminatorStateDeleting)
self.edgeClientConn.listener.factory.hostedServices.Delete(terminatorId)
if notifyCtrl {
logger.Info("removing terminator on controller")
ctrlCh := self.edgeClientConn.listener.factory.ctrls.AnyCtrlChannel()
if ctrlCh == nil {
logger.Error("no controller available, unable to remove terminator")
} else if err := self.edgeClientConn.removeTerminator(ctrlCh, self.token, terminatorId); err != nil {
logger.WithError(err).Error("failed to remove terminator")
} else {
logger.Info("Successfully removed terminator")
}
}
} else {
logger.Warn("edge terminator closing, but no terminator id set, so can't remove on controller")
}
}
} else {
if notifyCtrl {
if terminatorId := self.terminatorId.Load(); terminatorId != "" {
logger.Info("removing terminator on controller")
ctrlCh := self.edgeClientConn.listener.factory.ctrls.AnyCtrlChannel()
if ctrlCh == nil {
@@ -136,26 +178,12 @@ func (self *edgeTerminator) close(notify bool, reason string) {
} else if err := self.edgeClientConn.removeTerminator(ctrlCh, self.token, terminatorId); err != nil {
logger.WithError(err).Error("failed to remove terminator")
} else {
logger.Info("Successfully removed terminator")
logger.Info("successfully removed terminator")
}
} else {
logger.Warn("edge terminator closing, but no terminator id set, so can't remove on controller")
}
}
} else {
if terminatorId := self.terminatorId.Load(); terminatorId != "" {
logger.Info("removing terminator on controller")
ctrlCh := self.edgeClientConn.listener.factory.ctrls.AnyCtrlChannel()
if ctrlCh == nil {
logger.Error("no controller available, unable to remove terminator")
} else if err := self.edgeClientConn.removeTerminator(ctrlCh, self.token, terminatorId); err != nil {
logger.WithError(err).Error("failed to remove terminator")
} else {
logger.Info("successfully removed terminator")
}
} else {
logger.Warn("edge terminator closing, but no terminator id set, so can't remove on controller")
}
logger.Debug("removing terminator on router")
self.edgeClientConn.listener.factory.hostedServices.Delete(self.token)
router/xgress_edge/factory.go
+4 -1
View File
@@ -70,7 +70,10 @@ func (factory *Factory) BindChannel(binding channel.Binding) error {
binding.AddTypedReceiveHandler(handler_edge_ctrl.NewExtendEnrollmentCertsHandler(factory.env.GetRouterId(), func() {
factory.certChecker.CertsUpdated()
}))
binding.AddReceiveHandlerF(int32(edge_ctrl_pb.ContentType_CreateTerminatorV2ResponseType), factory.hostedServices.HandleCreateTerminatorResponse)
binding.AddTypedReceiveHandler(&channel.AsyncFunctionReceiveAdapter{
Type: int32(edge_ctrl_pb.ContentType_CreateTerminatorV2ResponseType),
Handler: factory.hostedServices.HandleCreateTerminatorResponse,
})
return nil
}
router/xgress_edge/hosted.go
+117 -120
View File
@@ -17,15 +17,14 @@
package xgress_edge
import (
"container/heap"
"fmt"
"github.com/cenkalti/backoff/v4"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel/v2"
"github.com/openziti/channel/v2/protobufs"
"github.com/openziti/sdk-golang/ziti/edge"
"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
routerEnv "github.com/openziti/ziti/router/env"
cmap "github.com/orcaman/concurrent-map/v2"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"google.golang.org/protobuf/proto"
@@ -35,22 +34,22 @@ import (
func newHostedServicesRegistry(env routerEnv.RouterEnv) *hostedServiceRegistry {
result := &hostedServiceRegistry{
services: sync.Map{},
events: make(chan terminatorEvent),
env: env,
retriesPending: false,
waits: cmap.New[chan struct{}](),
services: sync.Map{},
events: make(chan terminatorEvent),
env: env,
retriesPending: false,
terminatorQueue: &terminatorHeap{},
}
go result.run()
return result
}
type hostedServiceRegistry struct {
services sync.Map
events chan terminatorEvent
env routerEnv.RouterEnv
retriesPending bool
waits cmap.ConcurrentMap[string, chan struct{}]
services sync.Map
events chan terminatorEvent
env routerEnv.RouterEnv
retriesPending bool
terminatorQueue *terminatorHeap
}
type terminatorEvent interface {
@@ -58,13 +57,16 @@ type terminatorEvent interface {
}
func (self *hostedServiceRegistry) run() {
retryTicker := time.NewTicker(50 * time.Millisecond)
defer retryTicker.Stop()
queueCheckTicker := time.NewTicker(100 * time.Millisecond)
defer queueCheckTicker.Stop()
longQueueCheckTicker := time.NewTicker(time.Second)
defer longQueueCheckTicker.Stop()
for {
var retryChan <-chan time.Time
if self.retriesPending {
retryChan = retryTicker.C
retryChan = queueCheckTicker.C
}
select {
@@ -73,31 +75,66 @@ func (self *hostedServiceRegistry) run() {
case event := <-self.events:
event.handle(self)
case <-retryChan:
self.evaluateLinkStateQueue()
case <-longQueueCheckTicker.C:
self.scanForRetries()
}
}
}
func (self *hostedServiceRegistry) evaluateLinkStateQueue() {
now := time.Now()
for len(*self.terminatorQueue) > 0 {
next := (*self.terminatorQueue)[0]
if now.Before(next.nextAttempt) {
return
}
heap.Pop(self.terminatorQueue)
self.evaluateTerminator(next)
}
self.retriesPending = false
}
type establishTerminatorEvent struct {
terminator *edgeTerminator
}
func (self *establishTerminatorEvent) handle(registry *hostedServiceRegistry) {
registry.tryEstablish(self.terminator)
registry.evaluateTerminator(self.terminator)
}
type calculateRetry struct {
terminator *edgeTerminator
queueFailed bool
}
func (self *calculateRetry) handle(registry *hostedServiceRegistry) {
self.terminator.calculateRetry(self.queueFailed)
registry.retriesPending = true
}
func (self *hostedServiceRegistry) EstablishTerminator(terminator *edgeTerminator) {
event := &establishTerminatorEvent{
terminator: terminator,
}
self.Put(terminator.terminatorId.Load(), terminator)
self.queue(&establishTerminatorEvent{
terminator: terminator,
})
}
func (self *hostedServiceRegistry) queue(event terminatorEvent) {
select {
case <-self.env.GetCloseNotify():
pfxlog.Logger().WithField("terminatorId", terminator.terminatorId.Load()).
Error("unable to establish terminator, hosted service registry has been shutdown")
case self.events <- event:
case <-self.env.GetCloseNotify():
pfxlog.Logger().Error("unable to queue terminator event, hosted service registry has been shutdown")
}
}
func (self *hostedServiceRegistry) scheduleRetry(terminator *edgeTerminator, queueFailed bool) {
terminator.establishActive.Store(false)
if terminator.state.CompareAndSwap(TerminatorStateEstablished, TerminatorStatePendingEstablishment) {
self.queue(&calculateRetry{
terminator: terminator,
queueFailed: queueFailed,
})
}
}
@@ -105,31 +142,12 @@ func (self *hostedServiceRegistry) scanForRetries() {
self.services.Range(func(key, value any) bool {
terminator := value.(*edgeTerminator)
if terminator.state.Load() == TerminatorStatePendingEstablishment {
self.tryEstablish(terminator)
self.evaluateTerminator(terminator)
}
return true
})
}
func (self *hostedServiceRegistry) tryEstablish(terminator *edgeTerminator) {
log := pfxlog.Logger().WithField("terminatorId", terminator.Id()).
WithField("token", terminator.token).
WithField("state", terminator.state.Load().String())
if !terminator.state.CompareAndSwap(TerminatorStatePendingEstablishment, TerminatorStateEstablishing) {
log.Info("terminator not pending, not going to try to establish")
return
}
err := self.env.GetRateLimiterPool().QueueOrError(func() {
self.establishTerminatorWithRetry(terminator)
})
if err != nil {
log.Info("rate limited: unable to queue to establish")
self.retriesPending = true
}
}
func (self *hostedServiceRegistry) Put(hostId string, conn *edgeTerminator) {
self.services.Store(hostId, conn)
}
@@ -164,7 +182,7 @@ func (self *hostedServiceRegistry) cleanupServices(proxy *edgeClientConn) {
self.services.Range(func(key, value interface{}) bool {
terminator := value.(*edgeTerminator)
if terminator.edgeClientConn == proxy {
terminator.close(false, "") // don't notify, channel is already closed, we can't send messages
terminator.close(false, true, "channel closed") // don't notify, channel is already closed, we can't send messages
self.services.Delete(key)
}
return true
@@ -175,7 +193,7 @@ func (self *hostedServiceRegistry) cleanupDuplicates(newest *edgeTerminator) {
self.services.Range(func(key, value interface{}) bool {
terminator := value.(*edgeTerminator)
if terminator != newest && newest.token == terminator.token && newest.instance == terminator.instance {
terminator.close(false, "duplicate terminator") // don't notify, channel is already closed, we can't send messages
terminator.close(false, true, "duplicate terminator") // don't notify, channel is already closed, we can't send messages
self.services.Delete(key)
pfxlog.Logger().WithField("routerId", terminator.edgeClientConn.listener.id.Token).
WithField("token", terminator.token).
@@ -193,7 +211,7 @@ func (self *hostedServiceRegistry) unbindSession(connId uint32, sessionToken str
self.services.Range(func(key, value interface{}) bool {
terminator := value.(*edgeTerminator)
if terminator.MsgChannel.Id() == connId && terminator.token == sessionToken && terminator.edgeClientConn == proxy {
terminator.close(false, "unbind successful") // don't notify, sdk asked us to unbind
terminator.close(false, true, "unbind successful") // don't notify, sdk asked us to unbind
self.services.Delete(key)
pfxlog.Logger().WithField("routerId", terminator.edgeClientConn.listener.id.Token).
WithField("token", sessionToken).
@@ -218,49 +236,45 @@ func (self *hostedServiceRegistry) getRelatedTerminators(sessionToken string, pr
return result
}
func (self *hostedServiceRegistry) establishTerminatorWithRetry(terminator *edgeTerminator) {
func (self *hostedServiceRegistry) evaluateTerminator(terminator *edgeTerminator) {
log := logrus.
WithField("terminatorId", terminator.terminatorId.Load()).
WithField("state", terminator.state.Load()).
WithField("token", terminator.token)
if state := terminator.state.Load(); state != TerminatorStateEstablishing {
log.WithField("state", state.String()).Info("not attempting to establish terminator, not in establishing state")
if terminator.edgeClientConn.ch.IsClosed() {
log.Info("terminator sdk channel closed, not trying to establish")
return
}
operation := func() error {
if terminator.edgeClientConn.ch.IsClosed() {
return backoff.Permanent(fmt.Errorf("edge link is closed, stopping terminator creation for terminator %s",
terminator.terminatorId.Load()))
}
if state := terminator.state.Load(); state != TerminatorStateEstablishing {
return backoff.Permanent(fmt.Errorf("terminator state is %v, stopping terminator creation for terminator %s",
state.String(), terminator.terminatorId.Load()))
}
if terminator.terminatorId.Load() == "" {
return backoff.Permanent(fmt.Errorf("terminator has been closed, stopping terminator creation"))
}
err := self.establishTerminator(terminator)
if err != nil && terminator.state.Load() != TerminatorStateEstablishing {
return backoff.Permanent(err)
}
return err
if terminator.terminatorId.Load() == "" {
log.Info("terminator has been closed, not trying to establish")
return
}
expBackoff := backoff.NewExponentialBackOff()
expBackoff.InitialInterval = 5 * time.Second
expBackoff.MaxInterval = 5 * time.Minute
tryEstablish := terminator.state.Load() == TerminatorStatePendingEstablishment && terminator.nextAttempt.Before(time.Now())
if err := backoff.Retry(operation, expBackoff); err != nil {
log.WithError(err).Error("stopping attempts to establish terminator, see error")
} else if terminator.postValidate {
if result, err := terminator.inspect(true); err != nil {
log.WithError(err).Error("error validating terminator after create")
} else if result.Type != edge.ConnTypeBind {
log.WithError(err).Error("terminator invalid in sdk after create, closed")
} else {
log.Info("terminator validated successfully")
if tryEstablish && terminator.establishActive.CompareAndSwap(false, true) {
if !terminator.state.CompareAndSwap(TerminatorStatePendingEstablishment, TerminatorStateEstablishing) {
log.Infof("terminator in state %s, not pending establishment, not queueing", terminator.state.Load())
return
}
log.Info("queuing terminator to send create")
err := self.env.GetRateLimiterPool().QueueOrError(func() {
defer func() {
self.scheduleRetry(terminator, false)
}()
if err := self.establishTerminator(terminator); err != nil {
log.WithError(err).Error("error establishing terminator")
}
})
if err != nil {
log.Info("rate limited: unable to queue to establish")
self.scheduleRetry(terminator, true)
}
}
}
@@ -299,18 +313,7 @@ func (self *hostedServiceRegistry) establishTerminator(terminator *edgeTerminato
log.Info("sending create terminator v2 request")
err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendAndWaitForWire(ctrlCh)
if err != nil {
return err
}
if self.waitForTerminatorCreated(terminatorId, 10*time.Second) {
return nil
}
// return an error to indicate that we need to check if a response has come back after the next interval,
// and if not, re-send
return errors.Errorf("timeout waiting for response to create terminator request for terminator %v", terminator.terminatorId.Load())
return protobufs.MarshalTyped(request).WithTimeout(timeout).SendAndWaitForWire(ctrlCh)
}
func (self *hostedServiceRegistry) HandleCreateTerminatorResponse(msg *channel.Message, _ channel.Channel) {
@@ -331,47 +334,41 @@ func (self *hostedServiceRegistry) HandleCreateTerminatorResponse(msg *channel.M
return
}
log = log.WithField("lifetime", time.Since(terminator.createTime))
if response.Result == edge_ctrl_pb.CreateTerminatorResult_FailedBusy {
log.Info("controller too busy to handle create terminator, retrying later")
return
}
if response.Result != edge_ctrl_pb.CreateTerminatorResult_Success {
terminator.close(true, response.Msg)
if terminator.establishCallback != nil {
terminator.establishCallback(false, response.Msg)
}
terminator.close(true, false, response.Msg)
return
}
if terminator.state.CompareAndSwap(TerminatorStateEstablishing, TerminatorStateEstablished) {
self.notifyTerminatorCreated(response.TerminatorId)
log.Info("received terminator created notification")
} else {
log.Info("received additional terminator created notification")
}
if terminator.notifyEstablished {
go func() {
notifyMsg := channel.NewMessage(edge.ContentTypeBindSuccess, nil)
notifyMsg.PutUint32Header(edge.ConnIdHeader, terminator.MsgChannel.Id())
if err := notifyMsg.WithTimeout(time.Second * 30).Send(terminator.MsgChannel.Channel); err != nil {
log.WithError(err).Error("failed to send bind success")
}
}()
isValid := true
if terminator.postValidate {
if result, err := terminator.inspect(true); err != nil {
log.WithError(err).Error("error validating terminator after create")
} else if result.Type != edge.ConnTypeBind {
log.WithError(err).Error("terminator invalid in sdk after create, closed")
isValid = false
} else {
log.Info("terminator validated successfully")
}
}
}
func (self *hostedServiceRegistry) waitForTerminatorCreated(id string, timeout time.Duration) bool {
notifyC := make(chan struct{})
defer self.waits.Remove(id)
self.waits.Set(id, notifyC)
select {
case <-notifyC:
return true
case <-time.After(timeout):
return false
}
}
func (self *hostedServiceRegistry) notifyTerminatorCreated(id string) {
notifyC, _ := self.waits.Get(id)
if notifyC != nil {
close(notifyC)
if isValid && terminator.establishCallback != nil {
terminator.establishCallback(true, "terminator established")
}
}
router/xgress_edge/listener.go
+106 -38
View File
@@ -19,6 +19,7 @@ package xgress_edge
import (
"encoding/binary"
"fmt"
"github.com/openziti/ziti/common/ctrl_msg"
"time"
"github.com/openziti/ziti/common/capabilities"
@@ -166,18 +167,15 @@ func (self *edgeClientConn) processConnect(req *channel.Message, ch channel.Chan
terminatorIdentity, _ := req.GetStringHeader(edge.TerminatorIdentityHeader)
request := &edge_ctrl_pb.CreateCircuitRequest{
request := &ctrl_msg.CreateCircuitRequest{
SessionToken: token,
Fingerprints: self.fingerprints.Prints(),
TerminatorInstanceId: terminatorIdentity,
PeerData: peerData,
}
response := &edge_ctrl_pb.CreateCircuitResponse{}
timeout := self.listener.options.Options.GetCircuitTimeout
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendForReply(ctrlCh)
if err = getResultOrFailure(responseMsg, err, response); err != nil {
response, err := self.sendCreateCircuitRequest(request, ctrlCh)
if err != nil {
log.WithError(err).Warn("failed to dial fabric")
self.sendStateClosedReply(err.Error(), req)
conn.close(false, "failed to dial fabric")
@@ -192,6 +190,58 @@ func (self *edgeClientConn) processConnect(req *channel.Message, ch channel.Chan
x.Start()
}
func (self *edgeClientConn) sendCreateCircuitRequest(req *ctrl_msg.CreateCircuitRequest, ctrlCh channel.Channel) (*ctrl_msg.CreateCircuitResponse, error) {
if capabilities.IsCapable(ctrlCh, capabilities.ControllerCreateCircuitV2) {
return self.sendCreateCircuitRequestV2(req, ctrlCh)
}
return self.sendCreateCircuitRequestV1(req, ctrlCh)
}
func (self *edgeClientConn) sendCreateCircuitRequestV1(req *ctrl_msg.CreateCircuitRequest, ctrlCh channel.Channel) (*ctrl_msg.CreateCircuitResponse, error) {
request := &edge_ctrl_pb.CreateCircuitRequest{
SessionToken: req.SessionToken,
Fingerprints: req.Fingerprints,
TerminatorInstanceId: req.TerminatorInstanceId,
PeerData: req.PeerData,
}
response := &edge_ctrl_pb.CreateCircuitResponse{}
timeout := self.listener.options.Options.GetCircuitTimeout
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendForReply(ctrlCh)
if err = getResultOrFailure(responseMsg, err, response); err != nil {
return nil, err
}
return &ctrl_msg.CreateCircuitResponse{
CircuitId: response.CircuitId,
Address: response.Address,
PeerData: response.PeerData,
Tags: response.Tags,
}, nil
}
func (self *edgeClientConn) sendCreateCircuitRequestV2(req *ctrl_msg.CreateCircuitRequest, ctrlCh channel.Channel) (*ctrl_msg.CreateCircuitResponse, error) {
timeout := self.listener.options.Options.GetCircuitTimeout
msg, err := req.ToMessage().WithTimeout(timeout).SendForReply(ctrlCh)
if err != nil {
return nil, err
}
if msg.ContentType == int32(edge_ctrl_pb.ContentType_ErrorType) {
msg := string(msg.Body)
if msg == "" {
msg = "error state returned from controller with no message"
}
return nil, errors.New(msg)
}
if msg.ContentType != int32(edge_ctrl_pb.ContentType_CreateCircuitV2ResponseType) {
return nil, errors.Errorf("unexpected response type %v to request. expected %v",
msg.ContentType, edge_ctrl_pb.ContentType_CreateCircuitV2ResponseType)
}
return ctrl_msg.DecodeCreateCircuitResponse(msg)
}
func (self *edgeClientConn) processBind(req *channel.Message, ch channel.Channel) {
ctrlCh := self.listener.factory.ctrls.AnyCtrlChannel()
if ctrlCh == nil {
@@ -260,11 +310,12 @@ func (self *edgeClientConn) processBindV1(req *channel.Message, ch channel.Chann
edgeClientConn: self,
token: token,
assignIds: assignIds,
createTime: time.Now(),
}
// need to remove session remove listener on close
messageSink.onClose = self.listener.factory.stateManager.AddEdgeSessionRemovedListener(token, func(token string) {
messageSink.close(true, "session ended")
messageSink.close(true, true, "session ended")
})
self.listener.factory.hostedServices.Put(token, messageSink)
@@ -289,7 +340,7 @@ func (self *edgeClientConn) processBindV1(req *channel.Message, ch channel.Chann
responseMsg, err := protobufs.MarshalTyped(request).WithTimeout(timeout).SendForReply(ctrlCh)
if err = xgress_common.CheckForFailureResult(responseMsg, err, edge_ctrl_pb.ContentType_CreateTerminatorResponseType); err != nil {
log.WithError(err).Warn("error creating terminator")
messageSink.close(false, "") // don't notify here, as we're notifying next line with a response
messageSink.close(false, false, "") // don't notify here, as we're notifying next line with a response
self.sendStateClosedReply(err.Error(), req)
return
}
@@ -300,7 +351,7 @@ func (self *edgeClientConn) processBindV1(req *channel.Message, ch channel.Chann
if messageSink.MsgChannel.IsClosed() {
log.Warn("edge channel closed while setting up terminator. cleaning up terminator now")
messageSink.close(false, "edge channel closed")
messageSink.close(false, true, "edge channel closed")
return
}
@@ -390,42 +441,59 @@ func (self *edgeClientConn) processBindV2(req *channel.Message, ch channel.Chann
notifyEstablished, _ := req.GetBoolHeader(edge.SupportsBindSuccessHeader)
terminator := &edgeTerminator{
MsgChannel: *edge.NewEdgeMsgChannel(self.ch, connId),
edgeClientConn: self,
token: token,
cost: cost,
precedence: precedence,
instance: terminatorInstance,
instanceSecret: terminatorInstanceSecret,
hostData: hostData,
assignIds: assignIds,
v2: true,
postValidate: postValidate,
notifyEstablished: notifyEstablished,
MsgChannel: *edge.NewEdgeMsgChannel(self.ch, connId),
edgeClientConn: self,
token: token,
cost: cost,
precedence: precedence,
instance: terminatorInstance,
instanceSecret: terminatorInstanceSecret,
hostData: hostData,
assignIds: assignIds,
v2: true,
postValidate: postValidate,
createTime: time.Now(),
}
terminator.establishCallback = func(ok bool, msg string) {
if ok {
self.sendStateConnectedReply(req, nil)
if notifyEstablished {
notifyMsg := channel.NewMessage(edge.ContentTypeBindSuccess, nil)
notifyMsg.PutUint32Header(edge.ConnIdHeader, terminator.MsgChannel.Id())
if err := notifyMsg.WithTimeout(time.Second * 30).Send(terminator.MsgChannel.Channel); err != nil {
log.WithError(err).Error("failed to send bind success")
}
}
} else {
self.sendStateClosedReply(msg, req)
}
}
terminator.terminatorId.Store(terminatorId)
terminator.state.Store(TerminatorStatePendingEstablishment)
// need to remove session remove listener on close
terminator.onClose = self.listener.factory.stateManager.AddEdgeSessionRemovedListener(token, func(token string) {
terminator.close(true, "session ended")
})
// If the session was recently removed, the call to AddEdgeSessionRemovedListener will have asynchronously closed
// the terminator
if self.listener.factory.stateManager.WasSessionRecentlyRemoved(token) {
log.Info("invalid session, not establishing terminator")
} else {
log.Info("establishing terminator")
self.sendStateConnectedReply(req, nil)
self.listener.factory.hostedServices.EstablishTerminator(terminator)
if listenerId == "" {
// only removed dupes with a scan if we don't have an sdk provided key
self.listener.factory.hostedServices.cleanupDuplicates(terminator)
}
terminator.establishCallback(false, "invalid session")
return
}
// need to remove session remove listener on close
terminator.onClose = self.listener.factory.stateManager.AddEdgeSessionRemovedListener(token, func(token string) {
terminator.close(true, true, "session ended")
})
log.Info("establishing terminator")
self.listener.factory.hostedServices.EstablishTerminator(terminator)
if listenerId == "" {
// only removed dupes with a scan if we don't have an sdk provided key
self.listener.factory.hostedServices.cleanupDuplicates(terminator)
}
}
func (self *edgeClientConn) processUnbind(req *channel.Message, _ channel.Channel) {
router/xgress_edge/terminator_heap.go
+45
View File
@@ -0,0 +1,45 @@
/*
(c) Copyright NetFoundry Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
https://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package xgress_edge
type terminatorHeap []*edgeTerminator
func (self terminatorHeap) Len() int {
return len(self)
}
func (self terminatorHeap) Less(i, j int) bool {
return self[i].nextAttempt.Before(self[j].nextAttempt)
}
func (self terminatorHeap) Swap(i, j int) {
tmp := self[i]
self[i] = self[j]
self[j] = tmp
}
func (self *terminatorHeap) Push(x any) {
*self = append(*self, x.(*edgeTerminator))
}
func (self *terminatorHeap) Pop() any {
old := *self
n := len(old)
pm := old[n-1]
*self = old[0 : n-1]
return pm
}
router/xgress_edge_tunnel/factory.go
+35 -6
View File
@@ -17,17 +17,18 @@
package xgress_edge_tunnel
import (
"fmt"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/channel/v2"
"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
"github.com/openziti/ziti/router/fabric"
"github.com/openziti/ziti/router/handler_edge_ctrl"
"github.com/openziti/ziti/router"
"github.com/openziti/ziti/router/env"
"github.com/openziti/ziti/router/xgress"
"github.com/openziti/foundation/v2/stringz"
"github.com/openziti/identity"
"github.com/openziti/metrics"
"github.com/openziti/ziti/common/pb/edge_ctrl_pb"
"github.com/openziti/ziti/router"
"github.com/openziti/ziti/router/env"
"github.com/openziti/ziti/router/fabric"
"github.com/openziti/ziti/router/handler_edge_ctrl"
"github.com/openziti/ziti/router/xgress"
"github.com/pkg/errors"
"strings"
"time"
@@ -107,6 +108,9 @@ func (self *Factory) CreateListener(optionsData xgress.OptionsData) (xgress.List
return nil, err
}
self.tunneler.listenOptions = options
pfxlog.Logger().Debugf("xgress edge tunnel options: %v", options.ToLoggableString())
return self.tunneler, nil
}
@@ -234,3 +238,28 @@ func (options *Options) load(data xgress.OptionsData) error {
return nil
}
func (options *Options) ToLoggableString() string {
buf := strings.Builder{}
buf.WriteString(fmt.Sprintf("mtu=%v\n", options.Mtu))
buf.WriteString(fmt.Sprintf("randomDrops=%v\n", options.RandomDrops))
buf.WriteString(fmt.Sprintf("drop1InN=%v\n", options.Drop1InN))
buf.WriteString(fmt.Sprintf("txQueueSize=%v\n", options.TxQueueSize))
buf.WriteString(fmt.Sprintf("txPortalStartSize=%v\n", options.TxPortalStartSize))
buf.WriteString(fmt.Sprintf("txPortalMaxSize=%v\n", options.TxPortalMaxSize))
buf.WriteString(fmt.Sprintf("txPortalMinSize=%v\n", options.TxPortalMinSize))
buf.WriteString(fmt.Sprintf("txPortalIncreaseThresh=%v\n", options.TxPortalIncreaseThresh))
buf.WriteString(fmt.Sprintf("txPortalIncreaseScale=%v\n", options.TxPortalIncreaseScale))
buf.WriteString(fmt.Sprintf("txPortalRetxThresh=%v\n", options.TxPortalRetxThresh))
buf.WriteString(fmt.Sprintf("txPortalRetxScale=%v\n", options.TxPortalRetxScale))
buf.WriteString(fmt.Sprintf("txPortalDupAckThresh=%v\n", options.TxPortalDupAckThresh))
buf.WriteString(fmt.Sprintf("txPortalDupAckScale=%v\n", options.TxPortalDupAckScale))
buf.WriteString(fmt.Sprintf("rxBufferSize=%v\n", options.RxBufferSize))
buf.WriteString(fmt.Sprintf("retxStartMs=%v\n", options.RetxStartMs))
buf.WriteString(fmt.Sprintf("retxScale=%v\n", options.RetxScale))
buf.WriteString(fmt.Sprintf("retxAddMs=%v\n", options.RetxAddMs))
buf.WriteString(fmt.Sprintf("maxCloseWait=%v\n", options.MaxCloseWait))
buf.WriteString(fmt.Sprintf("getCircuitTimeout=%v\n", options.GetCircuitTimeout))
return buf.String()
}
zititest/go.mod
+12 -12
View File
@@ -9,18 +9,18 @@ require (
github.com/Jeffail/gabs/v2 v2.7.0
github.com/go-openapi/runtime v0.27.1
github.com/google/go-cmp v0.6.0
github.com/google/uuid v1.5.0
github.com/google/uuid v1.6.0
github.com/gorilla/websocket v1.5.1
github.com/michaelquigley/pfxlog v0.6.10
github.com/openziti/agent v1.0.16
github.com/openziti/channel/v2 v2.0.117
github.com/openziti/channel/v2 v2.0.119
github.com/openziti/edge-api v0.26.10
github.com/openziti/fablab v0.5.42
github.com/openziti/foundation/v2 v2.0.36
github.com/openziti/identity v1.0.69
github.com/openziti/sdk-golang v0.22.21
github.com/openziti/storage v0.2.28
github.com/openziti/transport/v2 v2.0.121
github.com/openziti/foundation/v2 v2.0.37
github.com/openziti/identity v1.0.70
github.com/openziti/sdk-golang v0.22.28
github.com/openziti/storage v0.2.30
github.com/openziti/transport/v2 v2.0.122
github.com/openziti/ziti v0.28.3
github.com/pkg/errors v0.9.1
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
@@ -136,8 +136,8 @@ require (
github.com/opentracing/opentracing-go v1.2.0 // indirect
github.com/openziti/dilithium v0.3.3 // indirect
github.com/openziti/jwks v1.0.3 // indirect
github.com/openziti/metrics v1.2.43 // indirect
github.com/openziti/runzmd v1.0.37 // indirect
github.com/openziti/metrics v1.2.45 // indirect
github.com/openziti/runzmd v1.0.38 // indirect
github.com/openziti/secretstream v0.1.16 // indirect
github.com/openziti/x509-claims v1.0.3 // indirect
github.com/openziti/xweb/v2 v2.1.0 // indirect
@@ -156,7 +156,7 @@ require (
github.com/russross/blackfriday v1.6.0 // indirect
github.com/sagikazarmark/locafero v0.4.0 // indirect
github.com/sagikazarmark/slog-shim v0.1.0 // indirect
github.com/shirou/gopsutil/v3 v3.23.12 // indirect
github.com/shirou/gopsutil/v3 v3.24.1 // indirect
github.com/shoenig/go-m1cpu v0.1.6 // indirect
github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e // indirect
github.com/sourcegraph/conc v0.3.0 // indirect
@@ -190,8 +190,8 @@ require (
golang.org/x/mod v0.14.0 // indirect
golang.org/x/oauth2 v0.16.0 // indirect
golang.org/x/sync v0.6.0 // indirect
golang.org/x/sys v0.16.0 // indirect
golang.org/x/term v0.16.0 // indirect
golang.org/x/sys v0.17.0 // indirect
golang.org/x/term v0.17.0 // indirect
golang.org/x/text v0.14.0 // indirect
golang.org/x/tools v0.16.0 // indirect
google.golang.org/appengine v1.6.8 // indirect
zititest/go.sum
+24 -24
View File
@@ -327,8 +327,8 @@ github.com/google/pprof v0.0.0-20210122040257-d980be63207e/go.mod h1:kpwsk12EmLe
github.com/google/pprof v0.0.0-20210226084205-cbba55b83ad5/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
github.com/google/renameio v0.1.0/go.mod h1:KWCgfxg9yswjAJkECMjeO8J8rahYeXnNhOm40UhjYkI=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
github.com/googleapis/gax-go/v2 v2.0.3/go.mod h1:LLvjysVCY1JZeum8Z6l8qUty8fiNwE08qbEPm1M08qg=
github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg=
@@ -587,32 +587,32 @@ github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYr
github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8=
github.com/openziti/agent v1.0.16 h1:9Saji+8hFE1NpzP2XzDhsVJbCrDlhixoLHfOpFt5Z+U=
github.com/openziti/agent v1.0.16/go.mod h1:zfm53+PVWoGFzjGGgQdKby5749G6VRYHe+eQJmoVKy4=
github.com/openziti/channel/v2 v2.0.117 h1:9EUNJIA4NCosn+Uci0gYNiDMsGBX52Jd34wpziW+zPA=
github.com/openziti/channel/v2 v2.0.117/go.mod h1:RADBBOkbwpFMmbznU3bZmUsBB646seqZnQ7GPI6Chjg=
github.com/openziti/channel/v2 v2.0.119 h1:stfSrnDqoTi78LMvQA3+NSivHjQnRrYKrgij5NaOENI=
github.com/openziti/channel/v2 v2.0.119/go.mod h1:lSRJwqmbkE34DgXYEmUhVCzwcQcx65vZGE8nuBNK458=
github.com/openziti/dilithium v0.3.3 h1:PLgQ6PMNLSTzCFbX/h98cmudgz/cU6TmjdSv5NAPD8k=
github.com/openziti/dilithium v0.3.3/go.mod h1:vsCjI2AU/hon9e+dLhUFbCNGesJDj2ASgkySOcpmvjo=
github.com/openziti/edge-api v0.26.10 h1:LEDuJHZsExi0PBVO9iVuIdZWJ7eFo/i4TJhXoSFmfOU=
github.com/openziti/edge-api v0.26.10/go.mod h1:FQLjav9AfqxQYSL0xKPDZ/JWTSZXApkk7jM2/iczGXM=
github.com/openziti/fablab v0.5.42 h1:vENJKfEba2T4sSLwlKDL/IzBYfY8iHnhc4umf6IESiY=
github.com/openziti/fablab v0.5.42/go.mod h1:HDT06y1QX8kO8ZQrgHvZmJsvc8iRybESGtlDLDII4ks=
github.com/openziti/foundation/v2 v2.0.36 h1:ogEIvsWur8/9mUzf9NOB4hRUyx372Uy6AmnHRcurIkY=
github.com/openziti/foundation/v2 v2.0.36/go.mod h1:MdK2oAJSwo7iCfvVdG16ZGz47qP7nG97ovnqEdXW2kQ=
github.com/openziti/identity v1.0.69 h1:wNgQomnv8ar2S1wge9jQK1jpqE2virOKKG8GyfTiHMQ=
github.com/openziti/identity v1.0.69/go.mod h1:+7hQNAG5nUUdz0165OubWpMHkP/ZVJB3nv9HUXXi9H8=
github.com/openziti/foundation/v2 v2.0.37 h1:7pa4vWrlwllEoLXaK2rx91AffLQJ8k5pvc92oWANavA=
github.com/openziti/foundation/v2 v2.0.37/go.mod h1:2NxzCnJbMw35U9RrFcdEaiXdxIMfBHOUNPngpyhvKeY=
github.com/openziti/identity v1.0.70 h1:JNwtJHmIS0DcXookm2xuXyh4z92T1O21GQvuO8PmHWs=
github.com/openziti/identity v1.0.70/go.mod h1:jsKBL4G1BsmDSCIfhK4jha5B3Sevgy1jyZq0GtFKhSk=
github.com/openziti/jwks v1.0.3 h1:hf8wkb+Cg4nH/HM0KROFd7u+C3DkRVcFZJ7tDV+4icc=
github.com/openziti/jwks v1.0.3/go.mod h1:t4xxq8vlXGsPn29kiQVnZBBDDnEoOFqtJoHibkJunQQ=
github.com/openziti/metrics v1.2.43 h1:DSrmpLhoA45DlLVNdKOn2lBfCM0/r6wKz+3SDXe8X7Y=
github.com/openziti/metrics v1.2.43/go.mod h1:+RY4avT60Vbxb9wyfvRD0msrARyYCB5+heb8VIZzCm8=
github.com/openziti/runzmd v1.0.37 h1:qj2r9z4t7OAdmIXMdGbP9Su6TqA0bLdD2RMjJ71LRS0=
github.com/openziti/runzmd v1.0.37/go.mod h1:eKhqJsGoLeDHex/o5Mw6TcNJxlVljafSVm7ZU+bX5G8=
github.com/openziti/sdk-golang v0.22.21 h1:rj8gCq9SPcRpStPX0GUL8X2vkCQNUn+nk2WbyY82b/Y=
github.com/openziti/sdk-golang v0.22.21/go.mod h1:w3M9yAVnW7IAU1dpYtiFB/mIYLiU7l4tB+c4coqWRwo=
github.com/openziti/metrics v1.2.45 h1:+3zqszLWyFdTgzbsQD90V0yJcC9Ek77qKaIGMQXkAXs=
github.com/openziti/metrics v1.2.45/go.mod h1:g6CgAEbFes2UtdfGrsR8AKkuoBVL5dkU61843uQvllM=
github.com/openziti/runzmd v1.0.38 h1:0eFWAf7R9Thx99ue7Gql2dBavsqv4jldU8W5AIZJ8wo=
github.com/openziti/runzmd v1.0.38/go.mod h1:WiQi+YIXxZBH1bwjr+eo/e3ftfTLEABpN0i2QIhBI9w=
github.com/openziti/sdk-golang v0.22.28 h1:s159CT42dXug4GiJiN/kM6/ol+N2LFZ2tUk6bOpbgiI=
github.com/openziti/sdk-golang v0.22.28/go.mod h1:BLaLvcLqAgf3JFoDPWLTj3j3X5rndo6ZejdDdkMlihQ=
github.com/openziti/secretstream v0.1.16 h1:tVanF7OpJL1MJ1gvWaRlR2i+kAbrGsxr3q6EXFOS08U=
github.com/openziti/secretstream v0.1.16/go.mod h1:bvjGBUW/0e5MzD5S3FW3rhGASRNWAi+kTkTENZ9qRDE=
github.com/openziti/storage v0.2.28 h1:qHnsSF4RgQpT23hOXlwkAvJ0gO5PLmvTqyBy13dm3Rc=
github.com/openziti/storage v0.2.28/go.mod h1:ahdvsmmdQWFsZXeExvXeA8MHUXPLrwuDFBJDTztLY5E=
github.com/openziti/transport/v2 v2.0.121 h1:9WB1/9YMFed2UuefqZq+odUYYkoFOr+8aBIAZEC1MNE=
github.com/openziti/transport/v2 v2.0.121/go.mod h1:hS7TzoyTIwm4lVEsGyjQkoEg+vA+6dAOR8S4FZmWIj4=
github.com/openziti/storage v0.2.30 h1:15o8rSSgtcNCSBONt81ZRASQXJHh2L9Y0aWcHoQ81dE=
github.com/openziti/storage v0.2.30/go.mod h1:ltsgk2G0c3jVZVTK5bHkRtblGdFFoSJ3fOMur/0iFKg=
github.com/openziti/transport/v2 v2.0.122 h1:XWwZ6JcSO1nvbZgfp6kdf8aR5LEEN343mpZlhSihirk=
github.com/openziti/transport/v2 v2.0.122/go.mod h1:07Ak2jMsyZmi7/ECxGNfMXk8cF1Vj5qtKj90FoAnK8A=
github.com/openziti/x509-claims v1.0.3 h1:HNdQ8Nf1agB3lBs1gahcO6zfkeS4S5xoQ2/PkY4HRX0=
github.com/openziti/x509-claims v1.0.3/go.mod h1:Z0WIpBm6c4ecrpRKrou6Gk2wrLWxJO/+tuUwKh8VewE=
github.com/openziti/xweb/v2 v2.1.0 h1:Xhh3C2pZkq/Prr65V+SfFSibLDYteoc4f62KQCcTZF4=
@@ -699,8 +699,8 @@ github.com/sagikazarmark/slog-shim v0.1.0 h1:diDBnUNK9N/354PgrxMywXnAwEr1QZcOr6g
github.com/sagikazarmark/slog-shim v0.1.0/go.mod h1:SrcSrq8aKtyuqEI1uvTDTK1arOWRIczQRv+GVI1AkeQ=
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
github.com/shirou/gopsutil/v3 v3.23.12 h1:z90NtUkp3bMtmICZKpC4+WaknU1eXtp5vtbQ11DgpE4=
github.com/shirou/gopsutil/v3 v3.23.12/go.mod h1:1FrWgea594Jp7qmjHUUPlJDTPgcsb9mGnXDxavtikzM=
github.com/shirou/gopsutil/v3 v3.24.1 h1:R3t6ondCEvmARp3wxODhXMTLC/klMa87h2PHUw5m7QI=
github.com/shirou/gopsutil/v3 v3.24.1/go.mod h1:UU7a2MSBQa+kW1uuDq8DeEBS8kmrnQwsv2b5O513rwU=
github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM=
github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ=
github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
@@ -1105,17 +1105,17 @@ golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.17.0 h1:25cE3gD+tdBA7lp7QfhuV+rJiE9YXTcS3VG1SqssI/Y=
golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
golang.org/x/term v0.13.0/go.mod h1:LTmsnFJwVN6bCy1rVCoS+qHT1HhALEFxKncY3WNNh4U=
golang.org/x/term v0.16.0 h1:m+B6fahuftsE9qjo0VWp2FW0mB3MTJvR0BaMQrq0pmE=
golang.org/x/term v0.16.0/go.mod h1:yn7UURbUtPyrVJPGPq404EukNFxcm/foM+bV/bfcDsY=
golang.org/x/term v0.17.0 h1:mkTF7LCd6WGJNL3K1Ad7kwxNfYAW6a8a8QqtMblp/4U=
golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk=
golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
zititest/models/links-test/main.go
+1
View File
@@ -224,6 +224,7 @@ var m = &model.Model{
workflow.AddAction(host.GroupExec("component.ctrl", 5, "rm -rf ./fablab/ctrldata"))
workflow.AddAction(component.Start(".ctrl"))
workflow.AddAction(semaphore.Sleep(2 * time.Second))
workflow.AddAction(edge.RaftJoin(".ctrl"))
workflow.AddAction(semaphore.Sleep(2 * time.Second))
workflow.AddAction(edge.InitRaftController("#ctrl1"))
zititest/models/zrok-test/main.go
+23 -3
View File
@@ -53,6 +53,7 @@ const TargetZrokVersion = ""
const iterations = 100_000
const pacing = 10 * time.Millisecond
const loopers = 2
//go:embed configs
var configResource embed.FS
@@ -66,7 +67,7 @@ func (self scaleStrategy) IsScaled(entity model.Entity) bool {
func (self scaleStrategy) GetEntityCount(entity model.Entity) uint32 {
if entity.GetType() == model.EntityTypeHost {
if entity.GetScope().HasTag("router") {
return 2
return 1
} else if entity.GetScope().HasTag("client") {
return 3
}
@@ -111,7 +112,7 @@ var m = &model.Model{
model.FactoryFunc(func(m *model.Model) error {
return m.ForEachHost("component.ctrl", 1, func(host *model.Host) error {
if host.InstanceType == "" {
host.InstanceType = "c5.large"
host.InstanceType = "c5.2xlarge"
}
return nil
})
@@ -194,6 +195,7 @@ var m = &model.Model{
Version: TargetZrokVersion,
Pacing: pacing,
Iterations: iterations,
Loopers: loopers,
},
},
},
@@ -225,6 +227,7 @@ var m = &model.Model{
Version: TargetZrokVersion,
Pacing: pacing,
Iterations: iterations,
Loopers: loopers,
},
},
},
@@ -256,6 +259,7 @@ var m = &model.Model{
Version: TargetZrokVersion,
Pacing: pacing,
Iterations: iterations,
Loopers: loopers,
},
},
},
@@ -288,6 +292,7 @@ var m = &model.Model{
Version: TargetZrokVersion,
Pacing: pacing,
Iterations: iterations,
Loopers: loopers,
},
},
},
@@ -323,13 +328,28 @@ var m = &model.Model{
workflow.AddAction(component.Start("#zrokFront"))
workflow.AddAction(semaphore.Sleep(2 * time.Second))
workflow.AddAction(component.ExecInParallelF(".client", 200, (*zitilab.ZrokLoopTestType).Init))
workflow.AddAction(component.ExecInParallelF(".client", 10, (*zitilab.ZrokLoopTestType).Init))
return workflow
}),
"clean": model.Bind(actions.Workflow(
component.StopInParallelHostExclusive("*", 15),
host.GroupExec("*", 25, "rm -f logs/*"),
)),
"startServers": model.Bind(actions.Workflow(
component.StopInParallelHostExclusive("*", 200),
component.Start(".ctrl"),
component.Start(".router"),
component.Start(".zrokCtrl"),
component.Start(".zrokFront"),
)),
"restartServers": model.Bind(actions.Workflow(
host.GroupExec("*", 50, "find fablab -type d -exec chmod 755 {} \\;"),
component.StopInParallelHostExclusive("*", 200),
component.Start(".ctrl"),
component.Start(".router"),
component.Start(".zrokCtrl"),
component.Start(".zrokFront"),
)),
"login": model.Bind(edge.Login("#ctrl1")),
},
zititest/zitilab/actions/edge/ctrl_init.go
+6 -3
View File
@@ -43,10 +43,13 @@ func (init *raftInit) Execute(run model.Run) error {
}
for _, c := range m.SelectComponents(init.componentSpec) {
sshConfigFactory := c.GetHost().NewSshConfigFactory()
ctrlType, ok := c.Type.(*zitilab.ControllerType)
if !ok {
return errors.Errorf("component %s is not a controller", c.Id)
}
tmpl := "set -o pipefail; /home/%s/fablab/bin/ziti agent controller init %s %s default.admin 2>&1 | tee logs/controller.edge.init.log"
if err := host.Exec(c.GetHost(), fmt.Sprintf(tmpl, sshConfigFactory.User(), username, password)).Execute(run); err != nil {
tmpl := "set -o pipefail; %s agent controller init %s %s default.admin 2>&1 | tee logs/controller.edge.init.log"
if err := host.Exec(c.GetHost(), fmt.Sprintf(tmpl, ctrlType.GetBinaryPath(c), username, password)).Execute(run); err != nil {
return err
}
}
zititest/zitilab/actions/edge/raft_join.go
+11 -4
View File
@@ -2,8 +2,9 @@ package edge
import (
"fmt"
"github.com/openziti/fablab/kernel/lib/actions/host"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/fablab/kernel/model"
"github.com/openziti/ziti/zititest/zitilab"
"github.com/pkg/errors"
)
@@ -23,10 +24,16 @@ func (self *raftJoin) Execute(run model.Run) error {
return errors.Errorf("no controllers found with spec '%v'", self.componentSpec)
}
primary := ctrls[0]
sshConfigFactory := primary.GetHost().NewSshConfigFactory()
ctrlType, ok := primary.Type.(*zitilab.ControllerType)
if !ok {
return errors.Errorf("component %s is not a controller", primary.Id)
}
log := pfxlog.Logger().WithField("component", primary.Id)
for _, c := range ctrls[1:] {
tmpl := "/home/%s/fablab/bin/ziti agent cluster add %v --id %v"
if err := host.Exec(primary.GetHost(), fmt.Sprintf(tmpl, sshConfigFactory.User(), "tls:"+c.Host.PublicIp+":6262", c.Id)).Execute(run); err != nil {
tmpl := "%s agent cluster add %v --id %v"
cmd := fmt.Sprintf(tmpl, ctrlType.GetBinaryPath(primary), "tls:"+c.Host.PublicIp+":6262", c.Id)
log.Info(cmd)
if err := primary.GetHost().ExecLogOnlyOnError(cmd); err != nil {
return err
}
}
zititest/zitilab/component_controller.go
+28 -5
View File
@@ -112,6 +112,10 @@ func (self *ControllerType) Stop(_ model.Run, c *model.Component) error {
return c.GetHost().KillProcesses("-TERM", self.getProcessFilter(c))
}
func (self *ControllerType) GetBinaryPath(c *model.Component) string {
return getZitiBinaryPath(c, self.Version)
}
func (self *ControllerType) InitStandalone(run model.Run, c *model.Component) error {
username := c.MustStringVariable("credentials.edge.username")
password := c.MustStringVariable("credentials.edge.password")
@@ -126,11 +130,6 @@ func (self *ControllerType) InitStandalone(run model.Run, c *model.Component) er
factory := c.GetHost().NewSshConfigFactory()
binaryName := "ziti"
if self.Version != "" {
binaryName += "-" + self.Version
}
binaryPath := getZitiBinaryPath(c, self.Version)
configPath := fmt.Sprintf("/home/%s/fablab/cfg/%s", factory.User(), self.getConfigName(c))
@@ -138,3 +137,27 @@ func (self *ControllerType) InitStandalone(run model.Run, c *model.Component) er
cmd := fmt.Sprintf(tmpl, factory.User(), binaryPath, configPath, username, password)
return host.Exec(c.GetHost(), cmd).Execute(run)
}
func (self *ControllerType) InitRaft(run model.Run, c *model.Component) error {
count := 0
err := run.GetModel().ForEachComponent("*", 1, func(other *model.Component) error {
if _, ok := c.Type.(*ControllerType); ok {
binaryPath := getZitiBinaryPath(c, self.Version)
tmpl := "%s agent cluster add %v --id %v"
cmd := fmt.Sprintf(tmpl, binaryPath, "tls:"+other.Host.PublicIp+":6262", other.Id)
count++
return c.GetHost().ExecLogOnlyOnError(cmd)
}
return nil
})
if err != nil {
return err
}
if count < 1 {
return errors.New("no other controllers found")
}
return nil
}
zititest/zitilab/component_zrok_controller.go
-2
View File
@@ -35,7 +35,6 @@ var _ model.FileStagingComponent = (*ZrokControllerType)(nil)
var _ model.ActionsComponent = (*ZrokControllerType)(nil)
const (
ZrokControllerActionInit = "init"
ZrokControllerActionPreCreateAccounts = "preCreateAccounts"
)
@@ -54,7 +53,6 @@ func (self *ZrokControllerType) InitType(*model.Component) {
func (self *ZrokControllerType) GetActions() map[string]model.ComponentAction {
return map[string]model.ComponentAction{
ZrokControllerActionInit: model.ComponentActionF(self.Init),
ZrokControllerActionPreCreateAccounts: model.ComponentActionF(self.PreCreateAccounts),
}
}
zititest/zitilab/component_zrok_frontend.go
-11
View File
@@ -35,11 +35,6 @@ import (
var _ model.ComponentType = (*ZrokFrontendType)(nil)
var _ model.ServerComponent = (*ZrokFrontendType)(nil)
var _ model.FileStagingComponent = (*ZrokFrontendType)(nil)
var _ model.ActionsComponent = (*ZrokFrontendType)(nil)
const (
ZrokFrontendActionInit = "init"
)
type ZrokFrontendType struct {
ConfigSourceFS fs.FS
@@ -58,12 +53,6 @@ func (self *ZrokFrontendType) InitType(*model.Component) {
}
}
func (self *ZrokFrontendType) GetActions() map[string]model.ComponentAction {
return map[string]model.ComponentAction{
ZrokFrontendActionInit: model.ComponentActionF(self.Init),
}
}
func (self *ZrokFrontendType) Dump() any {
return map[string]string{
"type_id": "zrok-frontend",
zititest/zitilab/component_zrok_looptest.go
+23 -24
View File
@@ -30,11 +30,6 @@ import (
var _ model.ComponentType = (*ZrokLoopTestType)(nil)
var _ model.ServerComponent = (*ZrokLoopTestType)(nil)
var _ model.FileStagingComponent = (*ZrokLoopTestType)(nil)
var _ model.ActionsComponent = (*ZrokLoopTestType)(nil)
const (
ZrokLoopTestActionInit = "init"
)
type ZrokLoopTestType struct {
Version string
@@ -54,12 +49,6 @@ func (self *ZrokLoopTestType) InitType(*model.Component) {
}
}
func (self *ZrokLoopTestType) GetActions() map[string]model.ComponentAction {
return map[string]model.ComponentAction{
ZrokFrontendActionInit: model.ComponentActionF(self.Init),
}
}
func (self *ZrokLoopTestType) Dump() any {
return map[string]string{
"type_id": "zrok-test-loop",
@@ -72,16 +61,17 @@ func (self *ZrokLoopTestType) StageFiles(r model.Run, c *model.Component) error
return stageziti.StageZrokOnce(r, c, self.Version, self.LocalPath)
}
func (self *ZrokLoopTestType) getProcessFilter() func(string) bool {
func (self *ZrokLoopTestType) getProcessFilter(c *model.Component) func(string) bool {
return func(s string) bool {
return strings.Contains(s, "zrok") &&
strings.Contains(s, " test loop public") &&
!strings.Contains(s, "sudo")
!strings.Contains(s, "sudo") &&
strings.Contains(s, fmt.Sprintf("--max-dwell-ms %d", 1000+c.ScaleIndex))
}
}
func (self *ZrokLoopTestType) IsRunning(_ model.Run, c *model.Component) (bool, error) {
pids, err := c.GetHost().FindProcesses(self.getProcessFilter())
pids, err := c.GetHost().FindProcesses(self.getProcessFilter(c))
if err != nil {
return false, err
}
@@ -95,8 +85,10 @@ func (self *ZrokLoopTestType) Start(_ model.Run, c *model.Component) error {
binaryPath := getBinaryPath(c, constants.ZROK, self.Version)
logsPath := fmt.Sprintf("/home/%s/logs/%s.log", user, c.Id)
serviceCmd := fmt.Sprintf("nohup sudo -u %s %s test loop public --iterations %v --loopers %v --min-pacing-ms %v --max-pacing-ms %v 2>&1 &> %s &",
userId, binaryPath, self.Iterations, self.Loopers, self.Pacing.Milliseconds(), self.Pacing.Milliseconds(), logsPath)
maxDwell := 1000 + c.ScaleIndex
serviceCmd := fmt.Sprintf("nohup sudo -u %s %s test loop public --iterations %v --loopers %v --min-pacing-ms %v --max-pacing-ms %v "+
"--max-dwell-ms %d 2>&1 &> %s &",
userId, binaryPath, self.Iterations, self.Loopers, self.Pacing.Milliseconds(), self.Pacing.Milliseconds(), maxDwell, logsPath)
if quiet, _ := c.GetBoolVariable("quiet_startup"); !quiet {
logrus.Info(serviceCmd)
@@ -115,21 +107,28 @@ func (self *ZrokLoopTestType) Start(_ model.Run, c *model.Component) error {
}
func (self *ZrokLoopTestType) Stop(_ model.Run, c *model.Component) error {
return c.GetHost().KillProcesses("-TERM", self.getProcessFilter())
return c.GetHost().KillProcesses("-TERM", self.getProcessFilter(c))
}
func (self *ZrokLoopTestType) getUnixUser(c *model.Component) string {
return fmt.Sprintf("zrok%v", c.ScaleIndex)
}
func (self *ZrokLoopTestType) Init(run model.Run, c *model.Component) error {
func (self *ZrokLoopTestType) InitializeHost(_ model.Run, c *model.Component) error {
userId := self.getUnixUser(c)
// this will error on first run
_ = c.GetHost().ExecLogOnlyOnError(fmt.Sprintf("sudo deluser %s --remove-home", userId))
if err := c.GetHost().ExecLogOnlyOnError(fmt.Sprintf("sudo useradd %s -m -g ubuntu ", userId)); err != nil {
return err
if _, err := c.GetHost().ExecLogged(fmt.Sprintf("id -u %s", userId)); err != nil {
cmd := fmt.Sprintf("sudo useradd %s -m -g ubuntu ", userId)
pfxlog.Logger().Info(cmd)
if err = c.GetHost().ExecLogOnlyOnError(fmt.Sprintf("sudo useradd %s -m -g ubuntu ", userId)); err != nil {
return err
}
}
return nil
}
func (self *ZrokLoopTestType) Init(run model.Run, c *model.Component) error {
userId := self.getUnixUser(c)
binaryPath := getBinaryPath(c, constants.ZROK, self.Version)
val, ok := c.Data["token"]
@@ -138,8 +137,8 @@ func (self *ZrokLoopTestType) Init(run model.Run, c *model.Component) error {
}
token := fmt.Sprintf("%v", val)
zrokApiEndpoint := run.GetModel().MustSelectHost("zrokCtrl").PublicIp + ":1280"
tmpl := "set -o pipefail; sudo -u %s ZROK_API_ENDPOINT=http://%s %s enable %s"
cmd := fmt.Sprintf(tmpl, userId, zrokApiEndpoint, binaryPath, token)
tmpl := "set -o pipefail; sudo -u %s rm -rf /home/%s/.zrok && sudo -u %s ZROK_API_ENDPOINT=http://%s %s enable --headless %s"
cmd := fmt.Sprintf(tmpl, userId, userId, userId, zrokApiEndpoint, binaryPath, token)
pfxlog.Logger().Info(cmd)
return c.GetHost().ExecLogOnlyOnError(cmd)
}