Author SHA1 Message Date
mats 06ee7b1a27 more data 2024-11-08 23:00:28 +01:00
mats 65a969443a stuff 2024-11-08 21:58:28 +01:00
mats dfcdf0939f update names 2024-11-08 14:12:15 +01:00
mats a2fd5ad62f update serializing 2024-11-08 14:11:26 +01:00
53 changed files with 635 additions and 7671 deletions
-30
View File
@@ -1,30 +0,0 @@
name: Build and Publish
run-name: ${{ gitea.actor }} is building 🚀
on: [push]
jobs:
BuildAndDeployAmd64:
runs-on: amd64
steps:
- name: Check out repository code
uses: actions/checkout@v4
- name: Build docker image
run: docker build --progress=plain -t registry.knatofs.se/go-cart-actor-amd64:latest .
- name: Push to registry
run: docker push registry.knatofs.se/go-cart-actor-amd64:latest
- name: Deploy to Kubernetes
run: kubectl apply -f deployment/deployment.yaml -n cart
- name: Rollout amd64 deployment
run: kubectl rollout restart deployment/cart-actor-x86 -n cart
BuildAndDeploy:
runs-on: arm64
steps:
- name: Check out repository code
uses: actions/checkout@v4
- name: Build docker image
run: docker build --progress=plain -t registry.knatofs.se/go-cart-actor .
- name: Push to registry
run: docker push registry.knatofs.se/go-cart-actor
- name: Rollout arm64 deployment
run: kubectl rollout restart deployment/cart-actor-arm64 -n cart
+1 -4
View File
@@ -1,4 +1 @@
__debug* __debug*
go-cart-actor
data/*.prot
data/*.go*
-17
View File
@@ -1,17 +0,0 @@
# syntax=docker/dockerfile:1
FROM golang:alpine AS build-stage
WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download
COPY proto ./proto
COPY *.go ./
RUN CGO_ENABLED=0 GOOS=linux go build -o /go-cart-actor
FROM gcr.io/distroless/base-debian11
WORKDIR /
COPY --from=build-stage /go-cart-actor /go-cart-actor
ENTRYPOINT ["/go-cart-actor"]
-396
View File
@@ -1,396 +0,0 @@
# gRPC Migration Plan
File: GRPC-MIGRATION-PLAN.md
Author: (Generated plan)
Status: Draft for review
Target Release: Next major version (breaking change no mixed compatibility)
---
## 1. Overview
This document describes the full migration of the current custom TCP frame-based protocol (both the cart mutation/state channel on port `1337` and the control plane on port `1338`) to gRPC. We will remove all legacy packet framing (`FrameWithPayload`, `RemoteGrain`, `GenericListener` handlers for these two ports) and replace them with two gRPC services:
1. Cart Actor Service (mutations + state retrieval)
2. Control Plane Service (cluster membership, negotiation, ownership change, lifecycle)
We intentionally keep:
- Internal `CartGrain` logic, message storage format, disk persistence, and JSON cart serialization.
- Existing message type numeric mapping for backward compatibility with persisted event logs.
- HTTP/REST API layer unchanged (it still consumes JSON state from the local/remote grain pipeline).
We do NOT implement mixed-version compatibility; migration occurs atomically (cluster restart with new image).
---
## 2. Goals
- Remove custom binary frame protocol & simplify maintenance.
- Provide clearer, strongly defined interfaces via `.proto` schemas.
- Improve observability via gRPC interceptors (metrics & tracing hooks).
- Reduce per-call overhead compared with the current manual connection pooling + handwritten framing (HTTP/2 multiplexing + connection reuse).
- Prepare groundwork for future enhancements (streaming, typed state, event streaming) without rewriting again.
---
## 3. Non-Goals (Phase 1)
- Converting the cart state payload from JSON to a strongly typed proto.
- Introducing authentication / mTLS (may be added later).
- Changing persistence or replay format.
- Changing the HTTP API contract.
- Implementing streaming watchers or push updates.
---
## 4. Architecture After Migration
Ports:
- `:1337` → gRPC CartActor service.
- `:1338` → gRPC ControlPlane service.
Each node:
- Runs one gRPC server with both services (can use a single listener bound to two services or keep two separate listeners; we will keep two ports initially to minimize operational surprise, but they could be merged later).
- Maintains a connection pool of `*grpc.ClientConn` objects keyed by remote hostname (one per remote host, reused for both services).
Call Flow (Mutation):
1. HTTP request hits `PoolServer`.
2. `SyncedPool.getGrain(cartId)`:
- Local: direct invocation.
- Remote: uses `RemoteGrainGRPC` (new) which invokes `CartActor.Mutate`.
3. Response JSON returned unchanged.
Control Plane Flow:
- Discovery (K8s watch) still triggers `AddRemote(host)`.
- Instead of custom `Ping`, `Negotiate`, etc. via frames, call gRPC methods on `ControlPlane` service.
- Ownership changes use `ConfirmOwner` RPC.
---
## 5. Proto Design
### 5.1 Cart Actor Proto (Envelope Pattern)
We keep an envelope with `bytes payload` holding the serialized underlying cart mutation proto (existing types in `messages.proto`). This minimizes churn.
Indented code block (proto sketch):
syntax = "proto3";
package cart;
option go_package = "git.tornberg.me/go-cart-actor/proto;proto";
enum MutationType {
MUTATION_TYPE_UNSPECIFIED = 0;
MUTATION_ADD_REQUEST = 1;
MUTATION_ADD_ITEM = 2;
MUTATION_REMOVE_ITEM = 4;
MUTATION_REMOVE_DELIVERY = 5;
MUTATION_CHANGE_QUANTITY = 6;
MUTATION_SET_DELIVERY = 7;
MUTATION_SET_PICKUP_POINT = 8;
MUTATION_CREATE_CHECKOUT_ORDER = 9;
MUTATION_SET_CART_ITEMS = 10;
MUTATION_ORDER_COMPLETED = 11;
}
message MutationRequest {
string cart_id = 1;
MutationType type = 2;
bytes payload = 3; // Serialized specific mutation proto
int64 client_timestamp = 4; // Optional; server fills if zero
}
message MutationReply {
int32 status_code = 1;
bytes payload = 2; // JSON cart state or error string
}
message StateRequest {
string cart_id = 1;
}
message StateReply {
int32 status_code = 1;
bytes payload = 2; // JSON cart state
}
service CartActor {
rpc Mutate(MutationRequest) returns (MutationReply);
rpc GetState(StateRequest) returns (StateReply);
}
### 5.2 Control Plane Proto
syntax = "proto3";
package control;
option go_package = "git.tornberg.me/go-cart-actor/proto;proto";
message Empty {}
message PingReply {
string host = 1;
int64 unix_time = 2;
}
message NegotiateRequest {
repeated string known_hosts = 1;
}
message NegotiateReply {
repeated string hosts = 1; // Healthy hosts returned
}
message CartIdsReply {
repeated string cart_ids = 1;
}
message OwnerChangeRequest {
string cart_id = 1;
string new_host = 2;
}
message OwnerChangeAck {
bool accepted = 1;
string message = 2;
}
message ClosingNotice {
string host = 1;
}
service ControlPlane {
rpc Ping(Empty) returns (PingReply);
rpc Negotiate(NegotiateRequest) returns (NegotiateReply);
rpc GetCartIds(Empty) returns (CartIdsReply);
rpc ConfirmOwner(OwnerChangeRequest) returns (OwnerChangeAck);
rpc Closing(ClosingNotice) returns (OwnerChangeAck);
}
---
## 6. Message Type Mapping
| Legacy Constant | Numeric | New Enum Value |
|-----------------|---------|-----------------------------|
| AddRequestType | 1 | MUTATION_ADD_REQUEST |
| AddItemType | 2 | MUTATION_ADD_ITEM |
| RemoveItemType | 4 | MUTATION_REMOVE_ITEM |
| RemoveDeliveryType | 5 | MUTATION_REMOVE_DELIVERY |
| ChangeQuantityType | 6 | MUTATION_CHANGE_QUANTITY |
| SetDeliveryType | 7 | MUTATION_SET_DELIVERY |
| SetPickupPointType | 8 | MUTATION_SET_PICKUP_POINT |
| CreateCheckoutOrderType | 9 | MUTATION_CREATE_CHECKOUT_ORDER |
| SetCartItemsType | 10 | MUTATION_SET_CART_ITEMS |
| OrderCompletedType | 11 | MUTATION_ORDER_COMPLETED |
Persisted events keep original numeric codes; reconstruction simply casts to `MutationType`.
---
## 7. Components To Remove / Replace
Remove (after migration complete):
- `remote-grain.go`
- `rpc-server.go`
- Any packet/frame-specific types solely used by the above (search: `FrameWithPayload`, `RemoteHandleMutation`, `RemoteGetState` where not reused by disk or internal logic).
- The constants representing network frame types in `synced-pool.go` (RemoteNegotiate, AckChange, etc.) replaced by gRPC calls.
- netpool usage for remote cart channel (control plane also no longer needs `Connection` abstraction).
Retain (until reworked or optionally cleaned later):
- `message.go` (for persistence)
- `message-handler.go`
- `cart-grain.go`
- `messages.proto` (underlying mutation messages)
- HTTP API server and REST handlers.
---
## 8. New / Modified Components
New files (planned):
- `proto/cart_actor.proto`
- `proto/control_plane.proto`
- `grpc/cart_actor_server.go` (server impl)
- `grpc/cart_actor_client.go` (client adapter implementing `Grain`)
- `grpc/control_plane_server.go`
- `grpc/control_plane_client.go`
- `grpc/interceptors.go` (metrics, logging, optional tracing hooks)
- `remote_grain_grpc.go` (adapter bridging existing interfaces)
- `control_plane_adapter.go` (replaces frame handlers in `SyncedPool`)
Modified:
- `synced-pool.go` (remote host management now uses gRPC clients; negotiation logic updated)
- `main.go` (initialize both gRPC services on startup)
- `go.mod` (add `google.golang.org/grpc`)
---
## 9. Step-by-Step Migration Plan
1. Add proto files and generate Go code (`protoc --go_out --go-grpc_out`).
2. Implement `CartActorServer`:
- Translate `MutationRequest` to `Message`.
- Use existing handler registry for payload encode/decode.
- Return JSON cart state.
3. Implement `CartActorClient` wrapper (`RemoteGrainGRPC`) implementing:
- `HandleMessage`: Build envelope, call `Mutate`.
- `GetCurrentState`: Call `GetState`.
4. Implement `ControlPlaneServer` with methods:
- `Ping`: returns host + time.
- `Negotiate`: merge host lists; emulate old logic.
- `GetCartIds`: iterate local grains.
- `ConfirmOwner`: replicate quorum flow (accept always; error path for future).
- `Closing`: schedule remote removal.
5. Implement `ControlPlaneClient` used inside `SyncedPool.AddRemote`.
6. Refactor `SyncedPool`:
- Replace frame handlers registration with gRPC client calls.
- Replace `Server.AddHandler(...)` start-up with launching gRPC server.
- Implement periodic health checks using `Ping`.
7. Remove old connection constructs for 1337/1338.
8. Metrics:
- Add unary interceptor capturing duration and status.
- Replace packet counters with `cart_grpc_mutate_calls_total`, `cart_grpc_control_calls_total`, histograms for latency.
9. Update `main.go` to start:
- gRPC server(s).
- HTTP server as before.
10. Delete legacy files & update README build instructions.
11. Load testing & profiling on Raspberry Pi hardware (or ARM emulation).
12. Final cleanup & dead code removal (search for now-unused constants & structs).
13. Tag release.
---
## 10. Performance Considerations (Raspberry Pi Focus)
- Single `*grpc.ClientConn` per remote host (HTTP/2 multiplexing) to reduce file descriptor and handshake overhead.
- Use small keepalive pings (optional) only if connections drop; default may suffice.
- Avoid reflection / dynamic dispatch in hot path: pre-build a mapping from `MutationType` to handler function.
- Reuse byte buffers:
- Implement a `sync.Pool` for mutation serialization to reduce GC pressure.
- Enforce per-RPC deadlines (e.g. 300400ms) to avoid pile-ups.
- Backpressure:
- Before dispatch: if local grain pool at capacity and target grain is remote, abort early with 503 to caller (optional).
- Disable gRPC compression for small payloads (mutation messages are small). Condition compression if payload > threshold (e.g. 8KB).
- Compile with `-ldflags="-s -w"` in production to reduce binary size (optional).
- Enable `GOMAXPROCS` tuned to CPU cores; Pi often benefits from leaving default but monitor.
- Use histograms with limited buckets to reduce Prometheus cardinality.
---
## 11. Testing Strategy
Unit:
- Message type mapping tests (legacy -> enum).
- Envelope roundtrip: Original proto -> payload -> gRPC -> server decode -> internal Message.
Integration:
- Two-node cluster simulation:
- Mutate cart on Node A, ownership moves, verify remote access from Node B.
- Quorum failure simulation (temporarily reject `ConfirmOwner`).
- Control plane negotiation: start nodes in staggered order, assert final membership.
Load/Perf:
- Benchmark local mutation vs remote mutation latency.
- High concurrency test (N goroutines each performing X mutations).
- Memory profiling (ensure no large buffer retention).
Failure Injection:
- Kill a node mid-mutation; client call should timeout and not corrupt local state.
- Simulated network partition: drop `Ping` replies; ensure host removal path triggers.
---
## 12. Rollback Strategy
Because no mixed-version compatibility is provided, rollback = redeploy previous version containing legacy protocol:
1. Stop all new-version pods.
2. Deploy old version cluster-wide.
3. No data migration needed (event persistence unaffected).
Note: Avoid partial upgrades; perform full rolling restart quickly to prevent split-brain (new nodes wont talk to old nodes).
---
## 13. Risks & Mitigations
| Risk | Description | Mitigation |
|------|-------------|------------|
| Full-cluster restart required | No mixed compatibility | Schedule maintenance window |
| gRPC adds CPU overhead | Envelope + marshaling cost | Buffer reuse, keep small messages uncompressed |
| Ownership race | Timing differences after refactor | Add explicit logs + tests around `RequestOwnership` path |
| Hidden dependency on frame-level status codes | Some code may assume `FrameWithPayload` fields | Wrap gRPC responses into minimal compatibility structs until fully removed |
| Memory growth | Connection reuse & pooled buffers not implemented initially | Add `sync.Pool` & track memory via pprof early |
---
## 14. Logging & Observability
- Structured log entries for:
- Ownership changes
- Negotiation rounds
- Remote spawn events
- Mutation failures (with cart id, mutation type)
- Metrics:
- `cart_grpc_mutate_duration_seconds` (histogram)
- `cart_grpc_mutate_errors_total`
- `cart_grpc_control_duration_seconds`
- `cart_remote_hosts` (gauge)
- Retain existing grain counts.
- Optional future: OpenTelemetry tracing (span per remote mutation).
---
## 15. Future Enhancements (Post-Migration)
- Replace JSON state with `CartState` proto and provide streaming watch API.
- mTLS between nodes (certificate rotation via K8s Secret or SPIRE).
- Distributed tracing integration.
- Ownership leasing with TTL and optimistic renewal.
- Delta replication or CRDT-based conflict resolution for experimentation.
---
## 16. Task Breakdown & Estimates
| Task | Estimate |
|------|----------|
| Proto definitions & generation | 0.5d |
| CartActor server/client | 1.0d |
| ControlPlane server/client | 1.0d |
| SyncedPool refactor | 1.0d |
| Metrics & interceptors | 0.5d |
| Remove legacy code & cleanup | 0.5d |
| Tests (unit + integration) | 1.5d |
| Benchmark & tuning | 0.51.0d |
| Total | ~67d |
---
## 17. Open Questions (Confirm Before Implementation)
1. Combine both services on a single port (simplify ops) or keep dual-port first? (Default here: keep dual, but easy to merge.)
2. Minimum Go version remains 1.24.x—acceptable to add `google.golang.org/grpc` latest?
3. Accept adding `sync.Pool` micro-optimizations in first pass or postpone?
---
## 18. Acceptance Criteria
- All previous integration tests (adjusted to gRPC) pass.
- Cart operations (add, remove, delivery, checkout) function across at least a 2node cluster.
- Control plane negotiation forms consistent host list.
- Latency for a remote mutation does not degrade beyond an acceptable threshold (define baseline before merge).
- Legacy networking code fully removed.
---
## 19. Next Steps (If Approved)
1. Implement proto files and commit.
2. Scaffold server & client code.
3. Refactor `SyncedPool` and `main.go`.
4. Add metrics and tests.
5. Run benchmark on target Pi hardware.
6. Review & merge.
---
End of Plan.
-178
View File
@@ -1,178 +0,0 @@
# Go Cart Actor
A distributed cart management system using the actor model pattern.
## Prerequisites
- Go 1.24.2+
- Protocol Buffers compiler (`protoc`)
- protoc-gen-go and protoc-gen-go-grpc plugins
### Installing Protocol Buffers
On Windows:
```powershell
winget install protobuf
```
On macOS:
```bash
brew install protobuf
```
On Linux:
```bash
# Ubuntu/Debian
sudo apt install protobuf-compiler
# Or download from: https://github.com/protocolbuffers/protobuf/releases
```
### Installing Go protobuf plugin
```bash
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
```
## Working with Protocol Buffers
### Generating Go code from proto files
After modifying any proto (`proto/messages.proto`, `proto/cart_actor.proto`, `proto/control_plane.proto`), regenerate the Go code (all three share the unified `messages` package):
```bash
cd proto
protoc --go_out=. --go_opt=paths=source_relative \
--go-grpc_out=. --go-grpc_opt=paths=source_relative \
messages.proto cart_actor.proto control_plane.proto
```
### Protocol Buffer Messages
The `proto/messages.proto` file defines the following message types:
- `AddRequest` - Add items to cart (includes quantity, sku, country, optional storeId)
- `SetCartRequest` - Set entire cart contents
- `AddItem` - Complete item information for cart
- `RemoveItem` - Remove item from cart
- `ChangeQuantity` - Update item quantity
- `SetDelivery` - Configure delivery options
- `SetPickupPoint` - Set pickup location
- `PickupPoint` - Pickup point details
- `RemoveDelivery` - Remove delivery option
- `CreateCheckoutOrder` - Initiate checkout
- `OrderCreated` - Order creation response
### Building the project
```bash
go build .
```
### Running tests
```bash
go test ./...
```
## HTTP API Quick Start (curl Examples)
Assuming the service is reachable at http://localhost:8080 and the cart API is mounted at /cart.
Most endpoints use an HTTP cookie named `cartid` to track the cart. The first request will set it.
### 1. Get (or create) a cart
```bash
curl -i http://localhost:8080/cart/
```
Response sets a `cartid` cookie and returns the current (possibly empty) cart JSON.
### 2. Add an item by SKU (implicit quantity = 1)
```bash
curl -i --cookie-jar cookies.txt http://localhost:8080/cart/add/TEST-SKU-123
```
Stores cookie in `cookies.txt` for subsequent calls.
### 3. Add an item with explicit payload (country, quantity)
```bash
curl -i --cookie cookies.txt \
-H "Content-Type: application/json" \
-d '{"sku":"TEST-SKU-456","quantity":2,"country":"se"}' \
http://localhost:8080/cart/
```
### 4. Change quantity of an existing line
(First list the cart to find `id` of the line; here we use id=1 as an example)
```bash
curl -i --cookie cookies.txt \
-X PUT -H "Content-Type: application/json" \
-d '{"id":1,"quantity":3}' \
http://localhost:8080/cart/
```
### 5. Remove an item
```bash
curl -i --cookie cookies.txt -X DELETE http://localhost:8080/cart/1
```
### 6. Set entire cart contents (overwrites items)
```bash
curl -i --cookie cookies.txt \
-X POST -H "Content-Type: application/json" \
-d '{"items":[{"sku":"TEST-SKU-AAA","quantity":1,"country":"se"},{"sku":"TEST-SKU-BBB","quantity":2,"country":"se"}]}' \
http://localhost:8080/cart/set
```
### 7. Add a delivery (provider + optional items)
If `items` is empty or omitted, all items without a delivery get this one.
```bash
curl -i --cookie cookies.txt \
-X POST -H "Content-Type: application/json" \
-d '{"provider":"standard","items":[1,2]}' \
http://localhost:8080/cart/delivery
```
### 8. Remove a delivery by deliveryId
```bash
curl -i --cookie cookies.txt -X DELETE http://localhost:8080/cart/delivery/1
```
### 9. Set a pickup point for a delivery
```bash
curl -i --cookie cookies.txt \
-X PUT -H "Content-Type: application/json" \
-d '{"id":"PUP123","name":"Locker 5","address":"Main St 1","city":"Stockholm","zip":"11122","country":"SE"}' \
http://localhost:8080/cart/delivery/1/pickupPoint
```
### 10. Checkout (returns HTML snippet from Klarna)
```bash
curl -i --cookie cookies.txt http://localhost:8080/cart/checkout
```
### 11. Using a known cart id directly (bypassing cookie)
If you already have a cart id (e.g. 1720000000000000):
```bash
CART_ID=1720000000000000
curl -i http://localhost:8080/cart/byid/$CART_ID
curl -i -X POST -H "Content-Type: application/json" \
-d '{"sku":"TEST-SKU-XYZ","quantity":1,"country":"se"}' \
http://localhost:8080/cart/byid/$CART_ID
```
### 12. Clear cart cookie (forces a new cart on next request)
```bash
curl -i --cookie cookies.txt -X DELETE http://localhost:8080/cart/
```
Tip: Use `--cookie-jar` and `--cookie` to persist the session across multiple commands:
```bash
curl --cookie-jar cookies.txt http://localhost:8080/cart/
curl --cookie cookies.txt http://localhost:8080/cart/add/TEST-SKU-123
```
## Important Notes
- Always regenerate protobuf Go code after modifying any `.proto` files (messages/cart_actor/control_plane)
- The generated `messages.pb.go` file should not be edited manually
- Make sure your PATH includes the protoc-gen-go binary location (usually `$GOPATH/bin`)
-83
View File
@@ -1,83 +0,0 @@
package main
import (
"log"
amqp "github.com/rabbitmq/amqp091-go"
)
type AmqpOrderHandler struct {
Url string
connection *amqp.Connection
//channel *amqp.Channel
}
const (
topic = "order-placed"
)
func (t *AmqpOrderHandler) Connect() error {
conn, err := amqp.DialConfig(t.Url, amqp.Config{
//Vhost: "/",
Properties: amqp.NewConnectionProperties(),
})
if err != nil {
return err
}
t.connection = conn
ch, err := conn.Channel()
if err != nil {
return err
}
defer ch.Close()
if err := ch.ExchangeDeclare(
topic, // name
"topic", // type
true, // durable
false, // auto-delete
false, // internal
false, // noWait
nil, // arguments
); err != nil {
return err
}
if _, err = ch.QueueDeclare(
topic, // name of the queue
true, // durable
false, // delete when unused
false, // exclusive
false, // noWait
nil, // arguments
); err != nil {
return err
}
return nil
}
func (t *AmqpOrderHandler) Close() error {
log.Println("Closing master channel")
return t.connection.Close()
//return t.channel.Close()
}
func (t *AmqpOrderHandler) OrderCompleted(data []byte) error {
ch, err := t.connection.Channel()
if err != nil {
return err
}
defer ch.Close()
return ch.Publish(
topic,
topic,
true,
false,
amqp.Publishing{
ContentType: "application/json",
Body: data,
},
)
}
-42
View File
@@ -1,42 +0,0 @@
### Add item to cart
POST https://cart.tornberg.me/api/12345
Content-Type: application/json
{
"sku": "763281",
"quantity": 1
}
### Update quanity of item in cart
PUT https://cart.tornberg.me/api/12345
Content-Type: application/json
{
"id": 1,
"quantity": 1
}
### Delete item from cart
DELETE https://cart.tornberg.me/api/1002/1
### Set delivery
POST https://cart.tornberg.me/api/1002/delivery
Content-Type: application/json
{
"provider": "postnord",
"items": []
}
### Get cart
GET https://cart.tornberg.me/api/12345
### Remove delivery method
DELETE https://cart.tornberg.me/api/12345/delivery/2
+54 -519
View File
@@ -1,35 +1,18 @@
package main package main
import ( import (
"bytes"
"encoding/json" "encoding/json"
"fmt" "fmt"
"log" "log"
"slices" "net/http"
"sync"
"time" "time"
messages "git.tornberg.me/go-cart-actor/proto" messages "git.tornberg.me/go-cart-actor/proto"
"github.com/matst80/slask-finder/pkg/index"
) )
type CartId [16]byte type CartId [16]byte
// String returns the cart id as a trimmed UTF-8 string (trailing zero bytes removed).
func (id CartId) String() string {
n := 0
for n < len(id) && id[n] != 0 {
n++
}
return string(id[:n])
}
// ToCartId converts an arbitrary string to a fixed-size CartId (truncating or padding with zeros).
func ToCartId(s string) CartId {
var id CartId
copy(id[:], []byte(s))
return id
}
func (id CartId) MarshalJSON() ([]byte, error) { func (id CartId) MarshalJSON() ([]byte, error) {
return json.Marshal(id.String()) return json.Marshal(id.String())
} }
@@ -44,72 +27,23 @@ func (id *CartId) UnmarshalJSON(data []byte) error {
return nil return nil
} }
type StockStatus int
const (
OutOfStock StockStatus = 0
LowStock StockStatus = 1
InStock StockStatus = 2
)
type CartItem struct { type CartItem struct {
Id int `json:"id"` Sku string `json:"sku"`
ItemId int `json:"itemId,omitempty"` Name string `json:"name"`
ParentId int `json:"parentId,omitempty"` Price int64 `json:"price"`
Sku string `json:"sku"` Image string `json:"image"`
Name string `json:"name"`
Price int64 `json:"price"`
TotalPrice int64 `json:"totalPrice"`
TotalTax int64 `json:"totalTax"`
OrgPrice int64 `json:"orgPrice"`
Stock StockStatus `json:"stock"`
Quantity int `json:"qty"`
Tax int `json:"tax"`
TaxRate int `json:"taxRate"`
Brand string `json:"brand,omitempty"`
Category string `json:"category,omitempty"`
Category2 string `json:"category2,omitempty"`
Category3 string `json:"category3,omitempty"`
Category4 string `json:"category4,omitempty"`
Category5 string `json:"category5,omitempty"`
Disclaimer string `json:"disclaimer,omitempty"`
SellerId string `json:"sellerId,omitempty"`
SellerName string `json:"sellerName,omitempty"`
ArticleType string `json:"type,omitempty"`
Image string `json:"image,omitempty"`
Outlet *string `json:"outlet,omitempty"`
StoreId *string `json:"storeId,omitempty"`
}
type CartDelivery struct {
Id int `json:"id"`
Provider string `json:"provider"`
Price int64 `json:"price"`
Items []int `json:"items"`
PickupPoint *messages.PickupPoint `json:"pickupPoint,omitempty"`
} }
type CartGrain struct { type CartGrain struct {
mu sync.RWMutex storageMessages []Message
lastItemId int Id CartId `json:"id"`
lastDeliveryId int Items []CartItem `json:"items"`
storageMessages []Message TotalPrice int64 `json:"totalPrice"`
Id CartId `json:"id"`
Items []*CartItem `json:"items"`
TotalPrice int64 `json:"totalPrice"`
TotalTax int64 `json:"totalTax"`
TotalDiscount int64 `json:"totalDiscount"`
Deliveries []*CartDelivery `json:"deliveries,omitempty"`
Processing bool `json:"processing"`
PaymentInProgress bool `json:"paymentInProgress"`
OrderReference string `json:"orderReference,omitempty"`
PaymentStatus string `json:"paymentStatus,omitempty"`
} }
type Grain interface { type Grain interface {
GetId() CartId GetId() CartId
HandleMessage(message *Message, isReplay bool) (*FrameWithPayload, error) HandleMessage(message *Message, isReplay bool) ([]byte, error)
GetCurrentState() (*FrameWithPayload, error)
} }
func (c *CartGrain) GetId() CartId { func (c *CartGrain) GetId() CartId {
@@ -123,90 +57,47 @@ func (c *CartGrain) GetLastChange() int64 {
return *c.storageMessages[len(c.storageMessages)-1].TimeStamp return *c.storageMessages[len(c.storageMessages)-1].TimeStamp
} }
func (c *CartGrain) GetCurrentState() (*FrameWithPayload, error) { func getItemData(sku string) (*messages.AddItem, error) {
result, err := json.Marshal(c) res, err := http.Get("https://slask-finder.tornberg.me/api/get/" + sku)
if err != nil {
ret := MakeFrameWithPayload(0, 400, []byte(err.Error()))
return &ret, nil
}
ret := MakeFrameWithPayload(0, 200, result)
return &ret, nil
}
func getInt(data float64, ok bool) (int, error) {
if !ok {
return 0, fmt.Errorf("invalid type")
}
return int(data), nil
}
func getItemData(sku string, qty int, country string) (*messages.AddItem, error) {
item, err := FetchItem(sku, country)
if err != nil { if err != nil {
return nil, err return nil, err
} }
orgPrice, _ := getInt(item.GetNumberFieldValue(5)) // getInt(item.Fields[5]) defer res.Body.Close()
var item index.DataItem
price, priceErr := getInt(item.GetNumberFieldValue(4)) //Fields[4] err = json.NewDecoder(res.Body).Decode(&item)
if err != nil {
if priceErr != nil { return nil, err
return nil, fmt.Errorf("invalid price")
} }
price := item.GetPrice()
if price == 0 {
priceField, ok := item.GetFields()[4]
if ok {
stock := InStock priceFloat, ok := priceField.(float64)
/*item.t if !ok {
if item.StockLevel == "0" || item.StockLevel == "" { price, ok = priceField.(int)
stock = OutOfStock if !ok {
} else if item.StockLevel == "5+" { return nil, fmt.Errorf("invalid price type")
stock = LowStock }
}*/ } else {
articleType, _ := item.GetStringFieldValue(1) //.Fields[1].(string) price = int(priceFloat)
outletGrade, ok := item.GetStringFieldValue(20) //.Fields[20].(string) }
var outlet *string }
if ok {
outlet = &outletGrade
} }
sellerId, _ := item.GetStringFieldValue(24) // .Fields[24].(string)
sellerName, _ := item.GetStringFieldValue(9) // .Fields[9].(string)
brand, _ := item.GetStringFieldValue(2) //.Fields[2].(string)
category, _ := item.GetStringFieldValue(10) //.Fields[10].(string)
category2, _ := item.GetStringFieldValue(11) //.Fields[11].(string)
category3, _ := item.GetStringFieldValue(12) //.Fields[12].(string)
category4, _ := item.GetStringFieldValue(13) //Fields[13].(string)
category5, _ := item.GetStringFieldValue(14) //.Fields[14].(string)
return &messages.AddItem{ return &messages.AddItem{
ItemId: int64(item.Id), Quantity: 1,
Quantity: int32(qty), Price: int64(price),
Price: int64(price), Sku: sku,
OrgPrice: int64(orgPrice), Name: item.Title,
Sku: sku, Image: item.Img,
Name: item.Title,
Image: item.Img,
Stock: int32(stock),
Brand: brand,
Category: category,
Category2: category2,
Category3: category3,
Category4: category4,
Category5: category5,
Tax: 2500,
SellerId: sellerId,
SellerName: sellerName,
ArticleType: articleType,
Disclaimer: item.Disclaimer,
Country: country,
Outlet: outlet,
}, nil }, nil
} }
func (c *CartGrain) AddItem(sku string, qty int, country string, storeId *string) (*FrameWithPayload, error) { func (c *CartGrain) AddItem(sku string) ([]byte, error) {
cartItem, err := getItemData(sku, qty, country) cartItem, err := getItemData(sku)
if err != nil { if err != nil {
return nil, err return nil, err
} }
cartItem.StoreId = storeId
return c.HandleMessage(&Message{ return c.HandleMessage(&Message{
Type: 2, Type: 2,
Content: cartItem, Content: cartItem,
@@ -214,10 +105,8 @@ func (c *CartGrain) AddItem(sku string, qty int, country string, storeId *string
} }
func (c *CartGrain) GetStorageMessage(since int64) []StorableMessage { func (c *CartGrain) GetStorageMessage(since int64) []StorableMessage {
c.mu.RLock()
defer c.mu.RUnlock()
ret := make([]StorableMessage, 0)
ret := make([]StorableMessage, 0)
for _, message := range c.storageMessages { for _, message := range c.storageMessages {
if *message.TimeStamp > since { if *message.TimeStamp > since {
ret = append(ret, message) ret = append(ret, message)
@@ -226,396 +115,42 @@ func (c *CartGrain) GetStorageMessage(since int64) []StorableMessage {
return ret return ret
} }
func (c *CartGrain) GetState() ([]byte, error) { func (c *CartGrain) HandleMessage(message *Message, isReplay bool) ([]byte, error) {
return json.Marshal(c) log.Printf("Handling message %d", message.Type)
}
func (c *CartGrain) ItemsWithDelivery() []int {
ret := make([]int, 0, len(c.Items))
for _, item := range c.Items {
for _, delivery := range c.Deliveries {
for _, id := range delivery.Items {
if item.Id == id {
ret = append(ret, id)
}
}
}
}
return ret
}
func (c *CartGrain) ItemsWithoutDelivery() []int {
ret := make([]int, 0, len(c.Items))
hasDelivery := c.ItemsWithDelivery()
for _, item := range c.Items {
found := false
for _, id := range hasDelivery {
if item.Id == id {
found = true
break
}
}
if !found {
ret = append(ret, item.Id)
}
}
return ret
}
func (c *CartGrain) FindItemWithSku(sku string) (*CartItem, bool) {
c.mu.RLock()
defer c.mu.RUnlock()
for _, item := range c.Items {
if item.Sku == sku {
return item, true
}
}
return nil, false
}
func GetTaxAmount(total int64, tax int) int64 {
taxD := 10000 / float64(tax)
return int64(float64(total) / float64((1 + taxD)))
}
func (c *CartGrain) HandleMessage(message *Message, isReplay bool) (*FrameWithPayload, error) {
if message.TimeStamp == nil { if message.TimeStamp == nil {
now := time.Now().Unix() now := time.Now().Unix()
message.TimeStamp = &now message.TimeStamp = &now
} }
grainMutations.Inc()
var err error var err error
switch message.Type { switch message.Type {
case SetCartItemsType:
msg, ok := message.Content.(*messages.SetCartRequest)
if !ok {
err = fmt.Errorf("expected SetCartItems")
} else {
c.mu.Lock()
c.Items = make([]*CartItem, 0, len(msg.Items))
c.mu.Unlock()
for _, item := range msg.Items {
c.AddItem(item.Sku, int(item.Quantity), item.Country, item.StoreId)
}
}
case AddRequestType: case AddRequestType:
msg, ok := message.Content.(*messages.AddRequest) msg, ok := message.Content.(*messages.AddRequest)
if !ok { if !ok {
err = fmt.Errorf("expected AddRequest") err = fmt.Errorf("invalid content type")
} else { } else {
return c.AddItem(msg.Sku)
existingItem, found := c.FindItemWithSku(msg.Sku)
if found {
existingItem.Quantity += int(msg.Quantity)
c.UpdateTotals()
} else {
return c.AddItem(msg.Sku, int(msg.Quantity), msg.Country, msg.StoreId)
}
} }
case AddItemType: case AddItemType:
msg, ok := message.Content.(*messages.AddItem) msg, ok := message.Content.(*messages.AddItem)
if !ok { if !ok {
err = fmt.Errorf("expected AddItem") err = fmt.Errorf("invalid content type")
} else { } else {
c.Items = append(c.Items, CartItem{
if msg.Quantity < 1 { Sku: msg.Sku,
return nil, fmt.Errorf("invalid quantity") Name: msg.Name,
} Price: msg.Price,
existingItem, found := c.FindItemWithSku(msg.Sku) Image: msg.Image,
if found { })
existingItem.Quantity += int(msg.Quantity) c.TotalPrice += msg.Price
c.UpdateTotals()
} else {
c.mu.Lock()
c.lastItemId++
tax := 2500
if msg.Tax > 0 {
tax = int(msg.Tax)
}
taxAmount := GetTaxAmount(msg.Price, tax)
c.Items = append(c.Items, &CartItem{
Id: c.lastItemId,
ItemId: int(msg.ItemId),
Quantity: int(msg.Quantity),
Sku: msg.Sku,
Name: msg.Name,
Price: msg.Price,
TotalPrice: msg.Price * int64(msg.Quantity),
TotalTax: int64(taxAmount * int64(msg.Quantity)),
Image: msg.Image,
Stock: StockStatus(msg.Stock),
Disclaimer: msg.Disclaimer,
Brand: msg.Brand,
Category: msg.Category,
Category2: msg.Category2,
Category3: msg.Category3,
Category4: msg.Category4,
Category5: msg.Category5,
OrgPrice: msg.OrgPrice,
ArticleType: msg.ArticleType,
Outlet: msg.Outlet,
SellerId: msg.SellerId,
SellerName: msg.SellerName,
Tax: int(taxAmount),
TaxRate: tax,
StoreId: msg.StoreId,
})
c.UpdateTotals()
c.mu.Unlock()
}
}
case ChangeQuantityType:
msg, ok := message.Content.(*messages.ChangeQuantity)
if !ok {
err = fmt.Errorf("expected ChangeQuantity")
} else {
for i, item := range c.Items {
if item.Id == int(msg.Id) {
if msg.Quantity <= 0 {
//c.TotalPrice -= item.Price * int64(item.Quantity)
c.Items = append(c.Items[:i], c.Items[i+1:]...)
} else {
//diff := int(msg.Quantity) - item.Quantity
item.Quantity = int(msg.Quantity)
//c.TotalPrice += item.Price * int64(diff)
}
break
}
}
c.UpdateTotals()
}
case RemoveItemType:
msg, ok := message.Content.(*messages.RemoveItem)
if !ok {
err = fmt.Errorf("expected RemoveItem")
} else {
items := make([]*CartItem, 0, len(c.Items))
for _, item := range c.Items {
if item.Id == int(msg.Id) {
//c.TotalPrice -= item.Price * int64(item.Quantity)
} else {
items = append(items, item)
}
}
c.Items = items
c.UpdateTotals()
}
case SetDeliveryType:
msg, ok := message.Content.(*messages.SetDelivery)
if !ok {
err = fmt.Errorf("expected SetDelivery")
} else {
c.lastDeliveryId++
items := make([]int, 0)
withDelivery := c.ItemsWithDelivery()
if len(msg.Items) == 0 {
items = append(items, c.ItemsWithoutDelivery()...)
} else {
for _, id := range msg.Items {
for _, item := range c.Items {
if item.Id == int(id) {
if slices.Contains(withDelivery, item.Id) {
return nil, fmt.Errorf("item already has delivery")
}
items = append(items, int(item.Id))
break
}
}
}
}
if len(items) > 0 {
c.Deliveries = append(c.Deliveries, &CartDelivery{
Id: c.lastDeliveryId,
Provider: msg.Provider,
PickupPoint: msg.PickupPoint,
Price: 4900,
Items: items,
})
c.UpdateTotals()
}
}
case RemoveDeliveryType:
msg, ok := message.Content.(*messages.RemoveDelivery)
if !ok {
err = fmt.Errorf("expected RemoveDelivery")
} else {
deliveries := make([]*CartDelivery, 0, len(c.Deliveries))
for _, delivery := range c.Deliveries {
if delivery.Id == int(msg.Id) {
c.TotalPrice -= delivery.Price
} else {
deliveries = append(deliveries, delivery)
}
}
c.Deliveries = deliveries
c.UpdateTotals()
}
case SetPickupPointType:
msg, ok := message.Content.(*messages.SetPickupPoint)
if !ok {
err = fmt.Errorf("expected SetPickupPoint")
} else {
for _, delivery := range c.Deliveries {
if delivery.Id == int(msg.DeliveryId) {
delivery.PickupPoint = &messages.PickupPoint{
Id: msg.Id,
Address: msg.Address,
City: msg.City,
Zip: msg.Zip,
Country: msg.Country,
Name: msg.Name,
}
break
}
}
}
case CreateCheckoutOrderType:
msg, ok := message.Content.(*messages.CreateCheckoutOrder)
if !ok {
err = fmt.Errorf("expected CreateCheckoutOrder")
} else {
orderLines := make([]*Line, 0, len(c.Items))
c.PaymentInProgress = true
c.Processing = true
for _, item := range c.Items {
orderLines = append(orderLines, &Line{
Type: "physical",
Reference: item.Sku,
Name: item.Name,
Quantity: item.Quantity,
UnitPrice: int(item.Price),
TaxRate: 2500, // item.TaxRate,
QuantityUnit: "st",
TotalAmount: int(item.TotalPrice),
TotalTaxAmount: int(item.TotalTax),
ImageURL: fmt.Sprintf("https://www.elgiganten.se%s", item.Image),
})
}
for _, line := range c.Deliveries {
if line.Price > 0 {
orderLines = append(orderLines, &Line{
Type: "shipping_fee",
Reference: line.Provider,
Name: "Delivery",
Quantity: 1,
UnitPrice: int(line.Price),
TaxRate: 2500, // item.TaxRate,
QuantityUnit: "st",
TotalAmount: int(line.Price),
TotalTaxAmount: int(GetTaxAmount(line.Price, 2500)),
})
}
}
order := CheckoutOrder{
PurchaseCountry: "SE",
PurchaseCurrency: "SEK",
Locale: "sv-se",
OrderAmount: int(c.TotalPrice),
OrderTaxAmount: int(c.TotalTax),
OrderLines: orderLines,
MerchantReference1: c.Id.String(),
MerchantURLS: &CheckoutMerchantURLS{
Terms: msg.Terms,
Checkout: msg.Checkout,
Confirmation: msg.Confirmation,
Validation: msg.Validation,
Push: msg.Push,
},
}
orderPayload, err := json.Marshal(order)
if err != nil {
return nil, err
}
var klarnaOrder *CheckoutOrder
if c.OrderReference != "" {
log.Printf("Updating order id %s", c.OrderReference)
klarnaOrder, err = KlarnaInstance.UpdateOrder(c.OrderReference, bytes.NewReader(orderPayload))
} else {
klarnaOrder, err = KlarnaInstance.CreateOrder(bytes.NewReader(orderPayload))
}
// log.Printf("Order result: %+v", klarnaOrder)
if nil != err {
log.Printf("error from klarna: %v", err)
return nil, err
}
if c.OrderReference == "" {
c.OrderReference = klarnaOrder.ID
c.PaymentStatus = klarnaOrder.Status
}
orderData, err := json.Marshal(klarnaOrder)
if nil != err {
return nil, err
}
result := MakeFrameWithPayload(RemoteCreateOrderReply, 200, orderData)
return &result, nil
}
case OrderCompletedType:
msg, ok := message.Content.(*messages.OrderCreated)
if !ok {
log.Printf("expected OrderCompleted, got %T", message.Content)
err = fmt.Errorf("expected OrderCompleted")
} else {
c.OrderReference = msg.OrderId
c.PaymentStatus = msg.Status
c.PaymentInProgress = false
} }
default: default:
err = fmt.Errorf("unknown message type %d", message.Type) err = fmt.Errorf("unknown message type")
} }
if err != nil { if err != nil {
return nil, err return nil, err
} }
if !isReplay { if !isReplay {
c.mu.Lock()
c.storageMessages = append(c.storageMessages, *message) c.storageMessages = append(c.storageMessages, *message)
c.mu.Unlock()
} }
result, err := json.Marshal(c) return json.Marshal(c)
msg := MakeFrameWithPayload(RemoteHandleMutationReply, 200, result)
return &msg, err
}
func (c *CartGrain) UpdateTotals() {
c.TotalPrice = 0
c.TotalTax = 0
c.TotalDiscount = 0
for _, item := range c.Items {
rowTotal := item.Price * int64(item.Quantity)
rowTax := int64(item.Tax) * int64(item.Quantity)
item.TotalPrice = rowTotal
item.TotalTax = rowTax
c.TotalPrice += rowTotal
c.TotalTax += rowTax
itemDiff := max(0, item.OrgPrice-item.Price)
c.TotalDiscount += itemDiff * int64(item.Quantity)
}
for _, delivery := range c.Deliveries {
c.TotalPrice += delivery.Price
c.TotalTax += GetTaxAmount(delivery.Price, 2500)
}
} }
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
Binary file not shown.
-276
View File
@@ -1,276 +0,0 @@
apiVersion: v1
kind: Secret
metadata:
name: klarna-api-credentials
data:
username: ZjQzZDY3YjEtNzA2Yy00NTk2LTliNTgtYjg1YjU2NDEwZTUw
password: a2xhcm5hX3Rlc3RfYXBpX0trUWhWVE5yYVZsV2FsTnhTRVp3Y1ZSSFF5UkVNRmxyY25Kd1AxSndQMGdzWmpRelpEWTNZakV0TnpBMll5MDBOVGsyTFRsaU5UZ3RZamcxWWpVMk5ERXdaVFV3TERFc2JUUkNjRFpWU1RsTllsSk1aMlEyVEc4MmRVODNZMkozUlRaaFdEZDViV3AwYkhGV1JqTjVNQzlaYXow
type: Opaque
---
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: cart-actor
arch: amd64
name: cart-actor-x86
spec:
replicas: 0
selector:
matchLabels:
app: cart-actor
arch: amd64
template:
metadata:
labels:
app: cart-actor
actor-pool: cart
arch: amd64
spec:
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: kubernetes.io/arch
operator: NotIn
values:
- arm64
volumes:
- name: data
nfs:
path: /i-data/7a8af061/nfs/cart-actor-no
server: 10.10.1.10
imagePullSecrets:
- name: regcred
serviceAccountName: default
containers:
- image: registry.knatofs.se/go-cart-actor-amd64:latest
name: cart-actor-amd64
imagePullPolicy: Always
lifecycle:
preStop:
exec:
command: ["sleep", "15"]
ports:
- containerPort: 8080
name: web
- containerPort: 1234
name: echo
- containerPort: 1337
name: rpc
- containerPort: 1338
name: quorum
livenessProbe:
httpGet:
path: /livez
port: web
failureThreshold: 1
periodSeconds: 10
readinessProbe:
httpGet:
path: /readyz
port: web
failureThreshold: 2
initialDelaySeconds: 2
periodSeconds: 10
volumeMounts:
- mountPath: "/data"
name: data
resources:
limits:
memory: "768Mi"
requests:
memory: "70Mi"
cpu: "1200m"
env:
- name: TZ
value: "Europe/Stockholm"
- name: KLARNA_API_USERNAME
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: username
- name: KLARNA_API_PASSWORD
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: password
- name: POD_IP
valueFrom:
fieldRef:
fieldPath: status.podIP
- name: AMQP_URL
value: "amqp://admin:12bananer@rabbitmq.dev:5672/"
- name: BASE_URL
value: "https://s10n-no.tornberg.me"
- name: CART_BASE_URL
value: "https://cart-no.tornberg.me"
- name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
---
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: cart-actor
arch: arm64
name: cart-actor-arm64
spec:
replicas: 3
selector:
matchLabels:
app: cart-actor
arch: arm64
template:
metadata:
labels:
app: cart-actor
actor-pool: cart
arch: arm64
spec:
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: kubernetes.io/hostname
operator: NotIn
values:
- masterpi
- key: kubernetes.io/arch
operator: In
values:
- arm64
volumes:
- name: data
nfs:
path: /i-data/7a8af061/nfs/cart-actor-no
server: 10.10.1.10
imagePullSecrets:
- name: regcred
serviceAccountName: default
containers:
- image: registry.knatofs.se/go-cart-actor:latest
name: cart-actor-arm64
imagePullPolicy: Always
lifecycle:
preStop:
exec:
command: ["sleep", "15"]
ports:
- containerPort: 8080
name: web
- containerPort: 1234
name: echo
- containerPort: 1337
name: rpc
- containerPort: 1338
name: quorum
livenessProbe:
httpGet:
path: /livez
port: web
failureThreshold: 1
periodSeconds: 10
readinessProbe:
httpGet:
path: /readyz
port: web
failureThreshold: 2
initialDelaySeconds: 2
periodSeconds: 10
volumeMounts:
- mountPath: "/data"
name: data
resources:
limits:
memory: "768Mi"
requests:
memory: "70Mi"
cpu: "1200m"
env:
- name: TZ
value: "Europe/Stockholm"
- name: KLARNA_API_USERNAME
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: username
- name: KLARNA_API_PASSWORD
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: password
- name: POD_IP
valueFrom:
fieldRef:
fieldPath: status.podIP
- name: AMQP_URL
value: "amqp://admin:12bananer@rabbitmq.dev:5672/"
- name: BASE_URL
value: "https://s10n-no.tornberg.me"
- name: CART_BASE_URL
value: "https://cart-no.tornberg.me"
- name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
---
kind: Service
apiVersion: v1
metadata:
name: cart-echo
spec:
selector:
app: cart-actor
type: LoadBalancer
ports:
- name: echo
port: 1234
---
kind: Service
apiVersion: v1
metadata:
name: cart-actor
annotations:
prometheus.io/port: "8080"
prometheus.io/scrape: "true"
prometheus.io/path: "/metrics"
spec:
selector:
app: cart-actor
ports:
- name: web
port: 8080
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: cart-ingress
annotations:
cert-manager.io/cluster-issuer: letsencrypt-prod
# nginx.ingress.kubernetes.io/affinity: "cookie"
# nginx.ingress.kubernetes.io/session-cookie-name: "cart-affinity"
# nginx.ingress.kubernetes.io/session-cookie-expires: "172800"
# nginx.ingress.kubernetes.io/session-cookie-max-age: "172800"
nginx.ingress.kubernetes.io/proxy-body-size: 4m
spec:
ingressClassName: nginx
tls:
- hosts:
- cart-no.tornberg.me
secretName: cart-actor-no-tls-secret
rules:
- host: cart-no.tornberg.me
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: cart-actor
port:
number: 8080
-272
View File
@@ -1,272 +0,0 @@
apiVersion: v1
kind: Secret
metadata:
name: klarna-api-credentials
data:
username: ZjQzZDY3YjEtNzA2Yy00NTk2LTliNTgtYjg1YjU2NDEwZTUw
password: a2xhcm5hX3Rlc3RfYXBpX0trUWhWVE5yYVZsV2FsTnhTRVp3Y1ZSSFF5UkVNRmxyY25Kd1AxSndQMGdzWmpRelpEWTNZakV0TnpBMll5MDBOVGsyTFRsaU5UZ3RZamcxWWpVMk5ERXdaVFV3TERFc2JUUkNjRFpWU1RsTllsSk1aMlEyVEc4MmRVODNZMkozUlRaaFdEZDViV3AwYkhGV1JqTjVNQzlaYXow
type: Opaque
---
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: cart-actor
arch: amd64
name: cart-actor-x86
spec:
replicas: 0
selector:
matchLabels:
app: cart-actor
arch: amd64
template:
metadata:
labels:
app: cart-actor
actor-pool: cart
arch: amd64
spec:
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: kubernetes.io/arch
operator: NotIn
values:
- arm64
volumes:
- name: data
nfs:
path: /i-data/7a8af061/nfs/cart-actor
server: 10.10.1.10
imagePullSecrets:
- name: regcred
serviceAccountName: default
containers:
- image: registry.knatofs.se/go-cart-actor-amd64:latest
name: cart-actor-amd64
imagePullPolicy: Always
lifecycle:
preStop:
exec:
command: ["sleep", "15"]
ports:
- containerPort: 8080
name: web
- containerPort: 1234
name: echo
- containerPort: 1337
name: rpc
- containerPort: 1338
name: quorum
livenessProbe:
httpGet:
path: /livez
port: web
failureThreshold: 1
periodSeconds: 10
readinessProbe:
httpGet:
path: /readyz
port: web
failureThreshold: 2
initialDelaySeconds: 2
periodSeconds: 10
volumeMounts:
- mountPath: "/data"
name: data
resources:
limits:
memory: "768Mi"
requests:
memory: "70Mi"
cpu: "1200m"
env:
- name: TZ
value: "Europe/Stockholm"
- name: KLARNA_API_USERNAME
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: username
- name: KLARNA_API_PASSWORD
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: password
- name: POD_IP
valueFrom:
fieldRef:
fieldPath: status.podIP
- name: AMQP_URL
value: "amqp://admin:12bananer@rabbitmq.dev:5672/"
# - name: BASE_URL
# value: "https://s10n-no.tornberg.me"
- name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
---
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: cart-actor
arch: arm64
name: cart-actor-arm64
spec:
replicas: 0
selector:
matchLabels:
app: cart-actor
arch: arm64
template:
metadata:
labels:
app: cart-actor
actor-pool: cart
arch: arm64
spec:
affinity:
nodeAffinity:
requiredDuringSchedulingIgnoredDuringExecution:
nodeSelectorTerms:
- matchExpressions:
- key: kubernetes.io/hostname
operator: NotIn
values:
- masterpi
- key: kubernetes.io/arch
operator: In
values:
- arm64
volumes:
- name: data
nfs:
path: /i-data/7a8af061/nfs/cart-actor
server: 10.10.1.10
imagePullSecrets:
- name: regcred
serviceAccountName: default
containers:
- image: registry.knatofs.se/go-cart-actor:latest
name: cart-actor-arm64
imagePullPolicy: Always
lifecycle:
preStop:
exec:
command: ["sleep", "15"]
ports:
- containerPort: 8080
name: web
- containerPort: 1234
name: echo
- containerPort: 1337
name: rpc
- containerPort: 1338
name: quorum
livenessProbe:
httpGet:
path: /livez
port: web
failureThreshold: 1
periodSeconds: 10
readinessProbe:
httpGet:
path: /readyz
port: web
failureThreshold: 2
initialDelaySeconds: 2
periodSeconds: 10
volumeMounts:
- mountPath: "/data"
name: data
resources:
limits:
memory: "768Mi"
requests:
memory: "70Mi"
cpu: "1200m"
env:
- name: TZ
value: "Europe/Stockholm"
- name: KLARNA_API_USERNAME
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: username
- name: KLARNA_API_PASSWORD
valueFrom:
secretKeyRef:
name: klarna-api-credentials
key: password
- name: POD_IP
valueFrom:
fieldRef:
fieldPath: status.podIP
- name: AMQP_URL
value: "amqp://admin:12bananer@rabbitmq.dev:5672/"
# - name: BASE_URL
# value: "https://s10n-no.tornberg.me"
- name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
---
kind: Service
apiVersion: v1
metadata:
name: cart-echo
spec:
selector:
app: cart-actor
type: LoadBalancer
ports:
- name: echo
port: 1234
---
kind: Service
apiVersion: v1
metadata:
name: cart-actor
annotations:
prometheus.io/port: "8080"
prometheus.io/scrape: "true"
prometheus.io/path: "/metrics"
spec:
selector:
app: cart-actor
ports:
- name: web
port: 8080
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: cart-ingress
annotations:
cert-manager.io/cluster-issuer: letsencrypt-prod
# nginx.ingress.kubernetes.io/affinity: "cookie"
# nginx.ingress.kubernetes.io/session-cookie-name: "cart-affinity"
# nginx.ingress.kubernetes.io/session-cookie-expires: "172800"
# nginx.ingress.kubernetes.io/session-cookie-max-age: "172800"
nginx.ingress.kubernetes.io/proxy-body-size: 4m
spec:
ingressClassName: nginx
tls:
- hosts:
- cart.tornberg.me
secretName: cart-actor-tls-secret
rules:
- host: cart.tornberg.me
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: cart-actor
port:
number: 8080
-22
View File
@@ -1,22 +0,0 @@
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: cart-discovery
rules:
- apiGroups: [""] # "" indicates the core API group
resources: ["pods","services", "deployments"]
verbs: ["get", "watch", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: cart-discovery-binding
subjects:
- kind: ServiceAccount
name: default
namespace: cart
apiGroup: ""
roleRef:
kind: ClusterRole
name: cart-discovery
apiGroup: ""
-25
View File
@@ -1,25 +0,0 @@
apiVersion: autoscaling/v1
kind: HorizontalPodAutoscaler
metadata:
name: cart-scaler-amd
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: cart-actor-x86
minReplicas: 3
maxReplicas: 9
targetCPUUtilizationPercentage: 30
---
apiVersion: autoscaling/v1
kind: HorizontalPodAutoscaler
metadata:
name: cart-scaler-arm
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: cart-actor-arm64
minReplicas: 3
maxReplicas: 9
targetCPUUtilizationPercentage: 30
-84
View File
@@ -1,84 +0,0 @@
package main
import (
"fmt"
"log"
"net"
"sync"
"time"
)
type DiscardedHost struct {
Host string
Tries int
}
type DiscardedHostHandler struct {
mu sync.RWMutex
port int
hosts []*DiscardedHost
onConnection *func(string)
}
func (d *DiscardedHostHandler) run() {
for range time.Tick(time.Second) {
d.mu.RLock()
lst := make([]*DiscardedHost, 0, len(d.hosts))
for _, host := range d.hosts {
if host.Tries >= 0 && host.Tries < 5 {
go d.testConnection(host)
lst = append(lst, host)
} else {
if host.Tries > 0 {
log.Printf("Host %s discarded after %d tries", host.Host, host.Tries)
}
}
}
d.mu.RUnlock()
d.mu.Lock()
d.hosts = lst
d.mu.Unlock()
}
}
func (d *DiscardedHostHandler) testConnection(host *DiscardedHost) {
addr := fmt.Sprintf("%s:%d", host.Host, d.port)
conn, err := net.Dial("tcp", addr)
if err != nil {
host.Tries++
if host.Tries >= 5 {
// Exceeded retry threshold; will be dropped by run loop.
}
} else {
conn.Close()
if d.onConnection != nil {
fn := *d.onConnection
fn(host.Host)
}
}
}
func NewDiscardedHostHandler(port int) *DiscardedHostHandler {
ret := &DiscardedHostHandler{
hosts: make([]*DiscardedHost, 0),
port: port,
}
go ret.run()
return ret
}
func (d *DiscardedHostHandler) SetReconnectHandler(fn func(string)) {
d.onConnection = &fn
}
func (d *DiscardedHostHandler) AppendHost(host string) {
d.mu.Lock()
defer d.mu.Unlock()
log.Printf("Adding host %s to retry list", host)
d.hosts = append(d.hosts, &DiscardedHost{
Host: host,
Tries: 0,
})
}
-77
View File
@@ -1,77 +0,0 @@
package main
import (
"context"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/cache"
toolsWatch "k8s.io/client-go/tools/watch"
)
type Discovery interface {
Discover() ([]string, error)
Watch() (<-chan HostChange, error)
}
type K8sDiscovery struct {
ctx context.Context
client *kubernetes.Clientset
}
func (k *K8sDiscovery) Discover() ([]string, error) {
return k.DiscoverInNamespace("")
}
func (k *K8sDiscovery) DiscoverInNamespace(namespace string) ([]string, error) {
pods, err := k.client.CoreV1().Pods(namespace).List(k.ctx, metav1.ListOptions{
LabelSelector: "actor-pool=cart",
})
if err != nil {
return nil, err
}
hosts := make([]string, 0, len(pods.Items))
for _, pod := range pods.Items {
hosts = append(hosts, pod.Status.PodIP)
}
return hosts, nil
}
type HostChange struct {
Host string
Type watch.EventType
}
func (k *K8sDiscovery) Watch() (<-chan HostChange, error) {
timeout := int64(30)
watcherFn := func(options metav1.ListOptions) (watch.Interface, error) {
return k.client.CoreV1().Pods("").Watch(k.ctx, metav1.ListOptions{
LabelSelector: "actor-pool=cart",
TimeoutSeconds: &timeout,
})
}
watcher, err := toolsWatch.NewRetryWatcher("1", &cache.ListWatch{WatchFunc: watcherFn})
if err != nil {
return nil, err
}
ch := make(chan HostChange)
go func() {
for event := range watcher.ResultChan() {
pod := event.Object.(*v1.Pod)
ch <- HostChange{
Host: pod.Status.PodIP,
Type: event.Type,
}
}
}()
return ch, nil
}
func NewK8sDiscovery(client *kubernetes.Clientset) *K8sDiscovery {
return &K8sDiscovery{
ctx: context.Background(),
client: client,
}
}
-51
View File
@@ -1,51 +0,0 @@
package main
import (
"testing"
"time"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/clientcmd"
)
func TestDiscovery(t *testing.T) {
config, err := clientcmd.BuildConfigFromFlags("", "/home/mats/.kube/config")
if err != nil {
t.Errorf("Error building config: %v", err)
}
client, err := kubernetes.NewForConfig(config)
if err != nil {
t.Errorf("Error creating client: %v", err)
}
d := NewK8sDiscovery(client)
res, err := d.DiscoverInNamespace("")
if err != nil {
t.Errorf("Error discovering: %v", err)
}
if len(res) == 0 {
t.Errorf("Expected at least one host, got none")
}
}
func TestWatch(t *testing.T) {
config, err := clientcmd.BuildConfigFromFlags("", "/home/mats/.kube/config")
if err != nil {
t.Errorf("Error building config: %v", err)
}
client, err := kubernetes.NewForConfig(config)
if err != nil {
t.Errorf("Error creating client: %v", err)
}
d := NewK8sDiscovery(client)
ch, err := d.Watch()
if err != nil {
t.Errorf("Error watching: %v", err)
}
select {
case m := <-ch:
t.Logf("Received watch %v", m)
case <-time.After(5 * time.Second):
t.Errorf("Timeout waiting for watch")
}
}
+6 -7
View File
@@ -2,6 +2,7 @@ package main
import ( import (
"encoding/gob" "encoding/gob"
"errors"
"fmt" "fmt"
"log" "log"
"os" "os"
@@ -24,11 +25,10 @@ func NewDiskStorage(stateFile string) (*DiskStorage, error) {
} }
func saveMessages(messages []StorableMessage, id CartId) error { func saveMessages(messages []StorableMessage, id CartId) error {
log.Printf("%d messages to save for %s", len(messages), id)
if len(messages) == 0 { if len(messages) == 0 {
return nil return nil
} }
log.Printf("%d messages to save for grain id %s", len(messages), id)
var file *os.File var file *os.File
var err error var err error
path := getCartPath(id.String()) path := getCartPath(id.String())
@@ -54,19 +54,18 @@ func getCartPath(id string) string {
func loadMessages(grain Grain, id CartId) error { func loadMessages(grain Grain, id CartId) error {
var err error var err error
path := getCartPath(id.String()) path := getCartPath(id.String())
if _, err = os.Stat(path); errors.Is(err, os.ErrNotExist) {
return err
}
file, err := os.Open(path) file, err := os.Open(path)
if err != nil { if err != nil {
if os.IsNotExist(err) {
return nil
}
return err return err
} }
defer file.Close() defer file.Close()
for err == nil { for err == nil {
var msg Message var msg Message
err = ReadMessage(file, &msg) err = MessageFromReader(file, &msg)
if err == nil { if err == nil {
grain.HandleMessage(&msg, true) grain.HandleMessage(&msg, true)
} }
-102
View File
@@ -1,102 +0,0 @@
package main
// Minimal frame abstractions retained after removal of the legacy TCP/frame
// networking layer. These types remain only to avoid a wide cascading refactor
// across existing grain / pool logic that still constructs and passes
// FrameWithPayload objects internally.
//
// The original responsibilities this replaces:
// - Binary framing, checksums, network IO
// - Distinction between request / reply frame types
//
// What remains:
// - A light weight container (FrameWithPayload) used as an inprocess
// envelope for status code + typed marker + payload bytes (JSON or proto).
// - Message / status constants referenced in existing code paths.
//
// Recommended future cleanup (postmigration):
// - Remove FrameType entirely and replace with enumerated semantic results
// or error values.
// - Replace FrameWithPayload with a struct { Status int; Data []byte }.
// - Remove remote_* reply type branching once all callers rely on gRPC
// status + strongly typed responses.
//
// For now we keep this minimal surface to keep the gRPC migration focused.
type (
// FrameType is a symbolic identifier carried through existing code paths.
// No ordering or bit semantics are required anymore.
FrameType uint32
StatusCode uint32
)
type Frame struct {
Type FrameType
StatusCode StatusCode
Length uint32
// Checksum retained for compatibility; no longer validated.
Checksum uint32
}
// FrameWithPayload wraps a Frame with an opaque payload.
// Payload usually contains JSON encoded cart state or an error message.
type FrameWithPayload struct {
Frame
Payload []byte
}
// -----------------------------------------------------------------------------
// Legacy Frame Type Constants (minimal subset still referenced)
// -----------------------------------------------------------------------------
const (
RemoteGetState = FrameType(0x01)
RemoteHandleMutation = FrameType(0x02)
ResponseBody = FrameType(0x03) // (rarely used; kept for completeness)
RemoteGetStateReply = FrameType(0x04)
RemoteHandleMutationReply = FrameType(0x05)
RemoteCreateOrderReply = FrameType(0x06)
)
// MakeFrameWithPayload constructs an inprocess frame wrapper.
// Length & Checksum are filled for backward compatibility (no validation logic
// depends on the checksum anymore).
func MakeFrameWithPayload(msg FrameType, statusCode StatusCode, payload []byte) FrameWithPayload {
length := uint32(len(payload))
return FrameWithPayload{
Frame: Frame{
Type: msg,
StatusCode: statusCode,
Length: length,
Checksum: (uint32(msg) + uint32(statusCode) + length) / 8, // simple legacy formula
},
Payload: payload,
}
}
// Clone creates a shallow copy of the frame, duplicating the payload slice.
func (f *FrameWithPayload) Clone() *FrameWithPayload {
if f == nil {
return nil
}
cp := make([]byte, len(f.Payload))
copy(cp, f.Payload)
return &FrameWithPayload{
Frame: f.Frame,
Payload: cp,
}
}
// NewErrorFrame helper for creating an error frame with a textual payload.
func NewErrorFrame(msg FrameType, code StatusCode, err error) FrameWithPayload {
var b []byte
if err != nil {
b = []byte(err.Error())
}
return MakeFrameWithPayload(msg, code, b)
}
// IsSuccess returns true if the status code indicates success in the
// conventional HTTP style range (200299). This mirrors previous usage patterns.
func (f *FrameWithPayload) IsSuccess() bool {
return f != nil && f.StatusCode >= 200 && f.StatusCode < 300
}
+17 -64
View File
@@ -1,73 +1,26 @@
module git.tornberg.me/go-cart-actor module git.tornberg.me/go-cart-actor
go 1.25.1 go 1.23.0
toolchain go1.23.2
require ( require (
github.com/google/uuid v1.6.0 cloud.google.com/go/compute/metadata v0.3.0 // indirect
github.com/matst80/slask-finder v0.0.0-20251009175145-ce05aff5a548
github.com/prometheus/client_golang v1.23.2
github.com/rabbitmq/amqp091-go v1.10.0
google.golang.org/grpc v1.76.0
google.golang.org/protobuf v1.36.10
k8s.io/api v0.34.1
k8s.io/apimachinery v0.34.1
k8s.io/client-go v0.34.1
)
require (
github.com/RoaringBitmap/roaring/v2 v2.10.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect github.com/beorn7/perks v1.0.1 // indirect
github.com/bits-and-blooms/bitset v1.24.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
github.com/emicklei/go-restful/v3 v3.13.0 // indirect github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
github.com/fxamacker/cbor/v2 v2.9.0 // indirect
github.com/go-logr/logr v1.4.3 // indirect
github.com/go-openapi/jsonpointer v0.22.1 // indirect
github.com/go-openapi/jsonreference v0.21.2 // indirect
github.com/go-openapi/swag v0.25.1 // indirect
github.com/go-openapi/swag/cmdutils v0.25.1 // indirect
github.com/go-openapi/swag/conv v0.25.1 // indirect
github.com/go-openapi/swag/fileutils v0.25.1 // indirect
github.com/go-openapi/swag/jsonname v0.25.1 // indirect
github.com/go-openapi/swag/jsonutils v0.25.1 // indirect
github.com/go-openapi/swag/loading v0.25.1 // indirect
github.com/go-openapi/swag/mangling v0.25.1 // indirect
github.com/go-openapi/swag/netutils v0.25.1 // indirect
github.com/go-openapi/swag/stringutils v0.25.1 // indirect
github.com/go-openapi/swag/typeutils v0.25.1 // indirect
github.com/go-openapi/swag/yamlutils v0.25.1 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/google/gnostic-models v0.7.0 // indirect
github.com/google/go-cmp v0.7.0 // indirect
github.com/gorilla/schema v1.4.1 // indirect github.com/gorilla/schema v1.4.1 // indirect
github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/compress v1.17.9 // indirect
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/matst80/slask-finder v0.0.0-20241104074525-3365cb1531ac // indirect
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
github.com/mschoch/smat v0.2.0 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_golang v1.20.4 // indirect
github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/client_model v0.6.1 // indirect
github.com/prometheus/common v0.67.1 // indirect github.com/prometheus/common v0.55.0 // indirect
github.com/prometheus/procfs v0.17.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect
github.com/spf13/pflag v1.0.6 // indirect github.com/rabbitmq/amqp091-go v1.10.0 // indirect
github.com/x448/float16 v0.8.4 // indirect github.com/redis/go-redis/v9 v9.5.3 // indirect
go.yaml.in/yaml/v2 v2.4.3 // indirect golang.org/x/oauth2 v0.23.0 // indirect
go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/sys v0.22.0 // indirect
golang.org/x/net v0.46.0 // indirect google.golang.org/protobuf v1.34.2 // indirect
golang.org/x/oauth2 v0.32.0 // indirect
golang.org/x/sys v0.37.0 // indirect
golang.org/x/term v0.36.0 // indirect
golang.org/x/text v0.30.0 // indirect
golang.org/x/time v0.14.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff // indirect
gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
k8s.io/klog/v2 v2.130.1 // indirect
k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912 // indirect
k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 // indirect
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
sigs.k8s.io/randfill v1.0.0 // indirect
sigs.k8s.io/structured-merge-diff/v6 v6.3.0 // indirect
sigs.k8s.io/yaml v1.6.0 // indirect
) )
+26 -209
View File
@@ -1,219 +1,36 @@
github.com/RoaringBitmap/roaring/v2 v2.10.0 h1:HbJ8Cs71lfCJyvmSptxeMX2PtvOC8yonlU0GQcy2Ak0= cloud.google.com/go/compute/metadata v0.3.0 h1:Tz+eQXMEqDIKRsmY3cHTL6FVaynIjX2QxYC4trgAKZc=
github.com/RoaringBitmap/roaring/v2 v2.10.0/go.mod h1:FiJcsfkGje/nZBZgCu0ZxCPOKD/hVXDS2dXi7/eUFE0= cloud.google.com/go/compute/metadata v0.3.0/go.mod h1:zFmK7XCadkQkj6TtorcaGlCW1hT1fIilQDwofLpJ20k=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/bits-and-blooms/bitset v1.12.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
github.com/bits-and-blooms/bitset v1.24.0 h1:H4x4TuulnokZKvHLfzVRTHJfFfnHEeSYJizujEZvmAM=
github.com/bits-and-blooms/bitset v1.24.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keLg81eXfW3O+oY=
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
github.com/emicklei/go-restful/v3 v3.13.0 h1:C4Bl2xDndpU6nJ4bc1jXd+uTmYPVUwkD6bFY/oTyCes=
github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM=
github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/go-openapi/jsonpointer v0.22.1 h1:sHYI1He3b9NqJ4wXLoJDKmUmHkWy/L7rtEo92JUxBNk=
github.com/go-openapi/jsonpointer v0.22.1/go.mod h1:pQT9OsLkfz1yWoMgYFy4x3U5GY5nUlsOn1qSBH5MkCM=
github.com/go-openapi/jsonreference v0.21.2 h1:Wxjda4M/BBQllegefXrY/9aq1fxBA8sI5M/lFU6tSWU=
github.com/go-openapi/jsonreference v0.21.2/go.mod h1:pp3PEjIsJ9CZDGCNOyXIQxsNuroxm8FAJ/+quA0yKzQ=
github.com/go-openapi/swag v0.25.1 h1:6uwVsx+/OuvFVPqfQmOOPsqTcm5/GkBhNwLqIR916n8=
github.com/go-openapi/swag v0.25.1/go.mod h1:bzONdGlT0fkStgGPd3bhZf1MnuPkf2YAys6h+jZipOo=
github.com/go-openapi/swag/cmdutils v0.25.1 h1:nDke3nAFDArAa631aitksFGj2omusks88GF1VwdYqPY=
github.com/go-openapi/swag/cmdutils v0.25.1/go.mod h1:pdae/AFo6WxLl5L0rq87eRzVPm/XRHM3MoYgRMvG4A0=
github.com/go-openapi/swag/conv v0.25.1 h1:+9o8YUg6QuqqBM5X6rYL/p1dpWeZRhoIt9x7CCP+he0=
github.com/go-openapi/swag/conv v0.25.1/go.mod h1:Z1mFEGPfyIKPu0806khI3zF+/EUXde+fdeksUl2NiDs=
github.com/go-openapi/swag/fileutils v0.25.1 h1:rSRXapjQequt7kqalKXdcpIegIShhTPXx7yw0kek2uU=
github.com/go-openapi/swag/fileutils v0.25.1/go.mod h1:+NXtt5xNZZqmpIpjqcujqojGFek9/w55b3ecmOdtg8M=
github.com/go-openapi/swag/jsonname v0.25.1 h1:Sgx+qbwa4ej6AomWC6pEfXrA6uP2RkaNjA9BR8a1RJU=
github.com/go-openapi/swag/jsonname v0.25.1/go.mod h1:71Tekow6UOLBD3wS7XhdT98g5J5GR13NOTQ9/6Q11Zo=
github.com/go-openapi/swag/jsonutils v0.25.1 h1:AihLHaD0brrkJoMqEZOBNzTLnk81Kg9cWr+SPtxtgl8=
github.com/go-openapi/swag/jsonutils v0.25.1/go.mod h1:JpEkAjxQXpiaHmRO04N1zE4qbUEg3b7Udll7AMGTNOo=
github.com/go-openapi/swag/jsonutils/fixtures_test v0.25.1 h1:DSQGcdB6G0N9c/KhtpYc71PzzGEIc/fZ1no35x4/XBY=
github.com/go-openapi/swag/jsonutils/fixtures_test v0.25.1/go.mod h1:kjmweouyPwRUEYMSrbAidoLMGeJ5p6zdHi9BgZiqmsg=
github.com/go-openapi/swag/loading v0.25.1 h1:6OruqzjWoJyanZOim58iG2vj934TysYVptyaoXS24kw=
github.com/go-openapi/swag/loading v0.25.1/go.mod h1:xoIe2EG32NOYYbqxvXgPzne989bWvSNoWoyQVWEZicc=
github.com/go-openapi/swag/mangling v0.25.1 h1:XzILnLzhZPZNtmxKaz/2xIGPQsBsvmCjrJOWGNz/ync=
github.com/go-openapi/swag/mangling v0.25.1/go.mod h1:CdiMQ6pnfAgyQGSOIYnZkXvqhnnwOn997uXZMAd/7mQ=
github.com/go-openapi/swag/netutils v0.25.1 h1:2wFLYahe40tDUHfKT1GRC4rfa5T1B4GWZ+msEFA4Fl4=
github.com/go-openapi/swag/netutils v0.25.1/go.mod h1:CAkkvqnUJX8NV96tNhEQvKz8SQo2KF0f7LleiJwIeRE=
github.com/go-openapi/swag/stringutils v0.25.1 h1:Xasqgjvk30eUe8VKdmyzKtjkVjeiXx1Iz0zDfMNpPbw=
github.com/go-openapi/swag/stringutils v0.25.1/go.mod h1:JLdSAq5169HaiDUbTvArA2yQxmgn4D6h4A+4HqVvAYg=
github.com/go-openapi/swag/typeutils v0.25.1 h1:rD/9HsEQieewNt6/k+JBwkxuAHktFtH3I3ysiFZqukA=
github.com/go-openapi/swag/typeutils v0.25.1/go.mod h1:9McMC/oCdS4BKwk2shEB7x17P6HmMmA6dQRtAkSnNb8=
github.com/go-openapi/swag/yamlutils v0.25.1 h1:mry5ez8joJwzvMbaTGLhw8pXUnhDK91oSJLDPF1bmGk=
github.com/go-openapi/swag/yamlutils v0.25.1/go.mod h1:cm9ywbzncy3y6uPm/97ysW8+wZ09qsks+9RS8fLWKqg=
github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo=
github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db h1:097atOisP2aRj7vFgYQBbFN4U4JNXUNYpxael3UzMyo=
github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144=
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/gorilla/schema v1.4.1 h1:jUg5hUjCSDZpNGLuXQOgIWGdlgrIdYvgQ0wZtdK1M3E= github.com/gorilla/schema v1.4.1 h1:jUg5hUjCSDZpNGLuXQOgIWGdlgrIdYvgQ0wZtdK1M3E=
github.com/gorilla/schema v1.4.1/go.mod h1:Dg5SSm5PV60mhF2NFaTV1xuYYj8tV8NOPRo4FggUMnM= github.com/gorilla/schema v1.4.1/go.mod h1:Dg5SSm5PV60mhF2NFaTV1xuYYj8tV8NOPRo4FggUMnM=
github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/matst80/slask-finder v0.0.0-20241104074525-3365cb1531ac h1:zakA1ck6dY4mMUGoZWGCjV3YP/8TiPtdJYvvieC6v8U=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/matst80/slask-finder v0.0.0-20241104074525-3365cb1531ac/go.mod h1:GCLeU45b+BNgLly5XbeB0A+47ctctp2SVHZ3NlfZqzs=
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/matst80/slask-finder v0.0.0-20251009175145-ce05aff5a548 h1:lkxVz5lNPlU78El49cx1c3Rxo/bp7Gbzp2BjSRFgg1U=
github.com/matst80/slask-finder v0.0.0-20251009175145-ce05aff5a548/go.mod h1:k4lo5gFYb3AqrgftlraCnv95xvjB/w8udRqJQJ12mAE=
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8=
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
github.com/mschoch/smat v0.2.0 h1:8imxQsjDm8yFEAVBe7azKmKSgzSkZXDuKkSq9374khM=
github.com/mschoch/smat v0.2.0/go.mod h1:kc9mz7DoBKqDyiRL7VZN8KvXQMWeTaVnttLRXOlotKw=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/onsi/ginkgo/v2 v2.21.0 h1:7rg/4f3rB88pb5obDgNZrNHrQ4e6WpjonchcpuBRnZM= github.com/prometheus/client_golang v1.20.4 h1:Tgh3Yr67PaOv/uTqloMsCEdeuFTatm5zIq5+qNN23vI=
github.com/onsi/ginkgo/v2 v2.21.0/go.mod h1:7Du3c42kxCUegi0IImZ1wUQzMBVecgIHjR1C+NkhLQo= github.com/prometheus/client_golang v1.20.4/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
github.com/onsi/gomega v1.35.1 h1:Cwbd75ZBPxFSuZ6T+rN/WCb/gOc6YgFBXLlZLhC7Ds4= github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
github.com/onsi/gomega v1.35.1/go.mod h1:PvZbdDc8J6XJEpDK4HCuRBm8a6Fzp9/DmhC9C7yFlog= github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.67.1 h1:OTSON1P4DNxzTg4hmKCc37o4ZAZDv0cfXLkOt0oEowI=
github.com/prometheus/common v0.67.1/go.mod h1:RpmT9v35q2Y+lsieQsdOh5sXZ6ajUGC8NjZAmr8vb0Q=
github.com/prometheus/procfs v0.17.0 h1:FuLQ+05u4ZI+SS/w9+BWEM2TXiHKsUQ9TADiRH7DuK0=
github.com/prometheus/procfs v0.17.0/go.mod h1:oPQLaDAMRbA+u8H5Pbfq+dl3VDAvHxMUOVhe0wYB2zw=
github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw= github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw=
github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= github.com/redis/go-redis/v9 v9.5.3 h1:fOAp1/uJG+ZtcITgZOfYFmTKPE7n4Vclj1wZFgRciUU=
github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= github.com/redis/go-redis/v9 v9.5.3/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=
github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= golang.org/x/oauth2 v0.23.0 h1:PbgcYx2W7i4LvjJWEbf0ngHV6qJYr86PkAV3bXdLEbs=
github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= golang.org/x/oauth2 v0.23.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= golang.org/x/sys v0.22.0 h1:RI27ohtqKCnwULzJLqkv897zojh5/DwS/ENaMzUOaWI=
github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= golang.org/x/sys v0.22.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I=
go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE=
go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E=
go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI=
go.opentelemetry.io/otel/sdk v1.37.0/go.mod h1:VredYzxUvuo2q3WRcDnKDjbdvmO0sCzOvVAiY+yUkAg=
go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFhbjxHHspCPc=
go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps=
go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0=
go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8=
go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
golang.org/x/net v0.46.0 h1:giFlY12I07fugqwPuWJi68oOnpfqFnJIJzaIIm2JVV4=
golang.org/x/net v0.46.0/go.mod h1:Q9BGdFy1y4nkUwiLvT5qtyhAnEHgnQ/zd8PfU6nc210=
golang.org/x/oauth2 v0.32.0 h1:jsCblLleRMDrxMN29H3z/k1KliIvpLgCkE6R8FXXNgY=
golang.org/x/oauth2 v0.32.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
golang.org/x/term v0.36.0 h1:zMPR+aF8gfksFprF/Nc/rd1wRS1EI6nDBGyWAvDzx2Q=
golang.org/x/term v0.36.0/go.mod h1:Qu394IJq6V6dCBRgwqshf3mPF85AqzYEzofzRdZkWss=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k=
golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM=
golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
golang.org/x/tools v0.37.0 h1:DVSRzp7FwePZW356yEAChSdNcQo6Nsp+fex1SUW09lE=
golang.org/x/tools v0.37.0/go.mod h1:MBN5QPQtLMHVdvsbtarmTNukZDdgwdwlO5qGacAzF0w=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff h1:A90eA31Wq6HOMIQlLfzFwzqGKBTuaVztYu/g8sn+8Zc=
google.golang.org/genproto/googleapis/rpc v0.0.0-20251007200510-49b9836ed3ff/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
google.golang.org/grpc v1.76.0 h1:UnVkv1+uMLYXoIz6o7chp59WfQUYA2ex/BXQ9rHZu7A=
google.golang.org/grpc v1.76.0/go.mod h1:Ju12QI8M6iQJtbcsV+awF5a4hfJMLi4X0JLo94ULZ6c=
google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo=
gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M=
gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.0/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
k8s.io/api v0.34.1 h1:jC+153630BMdlFukegoEL8E/yT7aLyQkIVuwhmwDgJM=
k8s.io/api v0.34.1/go.mod h1:SB80FxFtXn5/gwzCoN6QCtPD7Vbu5w2n1S0J5gFfTYk=
k8s.io/apimachinery v0.34.1 h1:dTlxFls/eikpJxmAC7MVE8oOeP1zryV7iRyIjB0gky4=
k8s.io/apimachinery v0.34.1/go.mod h1:/GwIlEcWuTX9zKIg2mbw0LRFIsXwrfoVxn+ef0X13lw=
k8s.io/client-go v0.34.1 h1:ZUPJKgXsnKwVwmKKdPfw4tB58+7/Ik3CrjOEhsiZ7mY=
k8s.io/client-go v0.34.1/go.mod h1:kA8v0FP+tk6sZA0yKLRG67LWjqufAoSHA2xVGKw9Of8=
k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk=
k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912 h1:Y3gxNAuB0OBLImH611+UDZcmKS3g6CthxToOb37KgwE=
k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912/go.mod h1:kdmbQkyfwUagLfXIad1y2TdrjPFWp2Q89B3qkRwf/pQ=
k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 h1:SjGebBtkBqHFOli+05xYbK8YF1Dzkbzn+gDM4X9T4Ck=
k8s.io/utils v0.0.0-20251002143259-bc988d571ff4/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg=
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg=
sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU=
sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY=
sigs.k8s.io/structured-merge-diff/v6 v6.3.0 h1:jTijUJbW353oVOd9oTlifJqOGEkUw2jB/fXCbTiQEco=
sigs.k8s.io/structured-merge-diff/v6 v6.3.0/go.mod h1:M3W8sfWvn2HhQDIbGWj3S099YozAsymCo/wrT5ohRUE=
sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs=
sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4=
+18 -87
View File
@@ -1,43 +1,22 @@
package main package main
import ( import (
"encoding/json"
"fmt" "fmt"
"log" "log"
"sync"
"time" "time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
var (
poolGrains = promauto.NewGauge(prometheus.GaugeOpts{
Name: "cart_grains_in_pool",
Help: "The total number of grains in the pool",
})
poolSize = promauto.NewGauge(prometheus.GaugeOpts{
Name: "cart_pool_size",
Help: "The total number of mutations",
})
poolUsage = promauto.NewGauge(prometheus.GaugeOpts{
Name: "cart_grain_pool_usage",
Help: "The current usage of the grain pool",
})
) )
type GrainPool interface { type GrainPool interface {
Process(id CartId, messages ...Message) (*FrameWithPayload, error) Process(id CartId, messages ...Message) (interface{}, error)
Get(id CartId) (*FrameWithPayload, error) Get(id CartId) (Grain, error)
} }
type Ttl struct { type Ttl struct {
Expires time.Time Expires time.Time
Grain *CartGrain Item *CartGrain
} }
type GrainLocalPool struct { type GrainLocalPool struct {
mu sync.RWMutex
grains map[CartId]*CartGrain grains map[CartId]*CartGrain
expiry []Ttl expiry []Ttl
spawn func(id CartId) (*CartGrain, error) spawn func(id CartId) (*CartGrain, error)
@@ -46,7 +25,6 @@ type GrainLocalPool struct {
} }
func NewGrainLocalPool(size int, ttl time.Duration, spawn func(id CartId) (*CartGrain, error)) *GrainLocalPool { func NewGrainLocalPool(size int, ttl time.Duration, spawn func(id CartId) (*CartGrain, error)) *GrainLocalPool {
ret := &GrainLocalPool{ ret := &GrainLocalPool{
spawn: spawn, spawn: spawn,
grains: make(map[CartId]*CartGrain), grains: make(map[CartId]*CartGrain),
@@ -54,7 +32,6 @@ func NewGrainLocalPool(size int, ttl time.Duration, spawn func(id CartId) (*Cart
Ttl: ttl, Ttl: ttl,
PoolSize: size, PoolSize: size,
} }
cartPurge := time.NewTicker(time.Minute) cartPurge := time.NewTicker(time.Minute)
go func() { go func() {
<-cartPurge.C <-cartPurge.C
@@ -63,45 +40,20 @@ func NewGrainLocalPool(size int, ttl time.Duration, spawn func(id CartId) (*Cart
return ret return ret
} }
func (p *GrainLocalPool) SetAvailable(availableWithLastChangeUnix map[CartId]int64) {
p.mu.Lock()
defer p.mu.Unlock()
for id := range availableWithLastChangeUnix {
if _, ok := p.grains[id]; !ok {
p.grains[id] = nil
p.expiry = append(p.expiry, Ttl{
Expires: time.Now().Add(p.Ttl),
Grain: nil,
})
}
}
}
func (p *GrainLocalPool) Purge() { func (p *GrainLocalPool) Purge() {
lastChangeTime := time.Now().Add(-p.Ttl) lastChangeTime := time.Now().Add(-p.Ttl)
keepChanged := lastChangeTime.Unix() keepChanged := lastChangeTime.Unix()
p.mu.Lock()
defer p.mu.Unlock()
for i := 0; i < len(p.expiry); i++ { for i := 0; i < len(p.expiry); i++ {
item := p.expiry[i] item := p.expiry[i]
if item.Expires.Before(time.Now()) { if item.Expires.Before(time.Now()) {
if item.Grain.GetLastChange() > keepChanged { if item.Item.GetLastChange() > keepChanged {
log.Printf("Expired item %s changed, keeping", item.Grain.GetId()) log.Printf("Changed item %s expired, keeping", item.Item.GetId())
if i < len(p.expiry)-1 { p.expiry = append(p.expiry[:i], p.expiry[i+1:]...)
p.expiry = append(p.expiry[:i], p.expiry[i+1:]...) p.expiry = append(p.expiry, item)
p.expiry = append(p.expiry, item)
} else {
p.expiry = append(p.expiry[:i], item)
}
} else { } else {
log.Printf("Item %s expired", item.Grain.GetId()) log.Printf("Item %s expired", item.Item.GetId())
delete(p.grains, item.Grain.GetId()) delete(p.grains, item.Item.GetId())
if i < len(p.expiry)-1 { p.expiry = append(p.expiry[:i], p.expiry[i+1:]...)
p.expiry = append(p.expiry[:i], p.expiry[i+1:]...)
} else {
p.expiry = p.expiry[:i]
}
} }
} else { } else {
break break
@@ -115,54 +67,33 @@ func (p *GrainLocalPool) GetGrains() map[CartId]*CartGrain {
func (p *GrainLocalPool) GetGrain(id CartId) (*CartGrain, error) { func (p *GrainLocalPool) GetGrain(id CartId) (*CartGrain, error) {
var err error var err error
// p.mu.RLock()
// defer p.mu.RUnlock()
grain, ok := p.grains[id] grain, ok := p.grains[id]
grainLookups.Inc() if !ok {
if grain == nil || !ok {
if len(p.grains) >= p.PoolSize { if len(p.grains) >= p.PoolSize {
if p.expiry[0].Expires.Before(time.Now()) { if p.expiry[0].Expires.Before(time.Now()) {
delete(p.grains, p.expiry[0].Grain.GetId()) delete(p.grains, p.expiry[0].Item.GetId())
p.expiry = p.expiry[1:] p.expiry = p.expiry[1:]
} else { } else {
return nil, fmt.Errorf("pool is full") return nil, fmt.Errorf("pool is full")
} }
} }
grain, err = p.spawn(id) grain, err = p.spawn(id)
p.mu.Lock()
p.grains[id] = grain p.grains[id] = grain
p.mu.Unlock()
} }
go func() {
l := float64(len(p.grains))
ps := float64(p.PoolSize)
poolUsage.Set(l / ps)
poolGrains.Set(l)
poolSize.Set(ps)
}()
return grain, err return grain, err
} }
func (p *GrainLocalPool) Process(id CartId, messages ...Message) (*FrameWithPayload, error) { func (p *GrainLocalPool) Process(id CartId, messages ...Message) (interface{}, error) {
grain, err := p.GetGrain(id) grain, err := p.GetGrain(id)
var result *FrameWithPayload
if err == nil && grain != nil { if err == nil && grain != nil {
for _, message := range messages { for _, message := range messages {
result, err = grain.HandleMessage(&message, false) _, err = grain.HandleMessage(&message, false)
} }
} }
return result, err return grain, err
} }
func (p *GrainLocalPool) Get(id CartId) (*FrameWithPayload, error) { func (p *GrainLocalPool) Get(id CartId) (Grain, error) {
grain, err := p.GetGrain(id) return p.GetGrain(id)
if err != nil {
return nil, err
}
data, err := json.Marshal(grain)
if err != nil {
return nil, err
}
ret := MakeFrameWithPayload(0, 200, data)
return &ret, nil
} }
+23
View File
@@ -0,0 +1,23 @@
package main
import (
"fmt"
"net"
"net/rpc"
)
type GrainServer struct {
Host string
}
func NewServer(hostname string) *GrainServer {
return &GrainServer{
Host: hostname,
}
}
func (s *GrainServer) Start(port int, instance Grain) (net.Listener, error) {
rpc.Register(instance)
rpc.HandleHTTP()
return net.Listen("tcp", fmt.Sprintf(":%d", port))
}
-135
View File
@@ -1,135 +0,0 @@
package main
import (
"bytes"
"context"
"encoding/json"
"fmt"
"testing"
"time"
messages "git.tornberg.me/go-cart-actor/proto"
"google.golang.org/grpc"
)
// TestCartActorMutationAndState validates end-to-end gRPC mutation + state retrieval
// against a locally started gRPC server (single-node scenario).
// This test uses AddItemType directly to avoid hitting external product
// fetching logic (FetchItem) which would require network I/O.
func TestCartActorMutationAndState(t *testing.T) {
// Setup local grain pool + synced pool (no discovery, single host)
pool := NewGrainLocalPool(1024, time.Minute, spawn)
synced, err := NewSyncedPool(pool, "127.0.0.1", nil)
if err != nil {
t.Fatalf("NewSyncedPool error: %v", err)
}
// Start gRPC server (CartActor + ControlPlane) on :1337
grpcSrv, err := StartGRPCServer(":1337", pool, synced)
if err != nil {
t.Fatalf("StartGRPCServer error: %v", err)
}
defer grpcSrv.GracefulStop()
// Dial the local server
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
conn, err := grpc.DialContext(ctx, "127.0.0.1:1337",
grpc.WithInsecure(),
grpc.WithBlock(),
)
if err != nil {
t.Fatalf("grpc.Dial error: %v", err)
}
defer conn.Close()
cartClient := messages.NewCartActorClient(conn)
// Create a short cart id (<=16 chars so it fits into the fixed CartId 16-byte array cleanly)
cartID := fmt.Sprintf("cart-%d", time.Now().UnixNano())
// Build an AddItem payload (bypasses FetchItem to keep test deterministic)
addItem := &messages.AddItem{
ItemId: 1,
Quantity: 1,
Price: 1000,
OrgPrice: 1000,
Sku: "test-sku",
Name: "Test SKU",
Image: "/img.png",
Stock: 2, // InStock
Tax: 2500,
Country: "se",
}
// Marshal underlying mutation payload using the existing handler code path
handler, ok := Handlers[AddItemType]
if !ok {
t.Fatalf("Handler for AddItemType missing")
}
payloadData, err := getSerializedPayload(handler, AddItemType, addItem)
if err != nil {
t.Fatalf("serialize add item: %v", err)
}
// Issue Mutate RPC
mutResp, err := cartClient.Mutate(context.Background(), &messages.MutationRequest{
CartId: cartID,
Type: messages.MutationType(AddItemType),
Payload: payloadData,
ClientTimestamp: time.Now().Unix(),
})
if err != nil {
t.Fatalf("Mutate RPC error: %v", err)
}
if mutResp.StatusCode != 200 {
t.Fatalf("Mutate returned non-200 status: %d payload=%s", mutResp.StatusCode, string(mutResp.Payload))
}
// Decode cart state JSON and validate
state := &CartGrain{}
if err := json.Unmarshal(mutResp.Payload, state); err != nil {
t.Fatalf("Unmarshal mutate cart state: %v\nPayload: %s", err, string(mutResp.Payload))
}
if len(state.Items) != 1 {
t.Fatalf("Expected 1 item after mutation, got %d", len(state.Items))
}
if state.Items[0].Sku != "test-sku" {
t.Fatalf("Unexpected item SKU: %s", state.Items[0].Sku)
}
// Issue GetState RPC
getResp, err := cartClient.GetState(context.Background(), &messages.StateRequest{
CartId: cartID,
})
if err != nil {
t.Fatalf("GetState RPC error: %v", err)
}
if getResp.StatusCode != 200 {
t.Fatalf("GetState returned non-200 status: %d payload=%s", getResp.StatusCode, string(getResp.Payload))
}
state2 := &CartGrain{}
if err := json.Unmarshal(getResp.Payload, state2); err != nil {
t.Fatalf("Unmarshal get state: %v", err)
}
if len(state2.Items) != 1 {
t.Fatalf("Expected 1 item in GetState, got %d", len(state2.Items))
}
if state2.Items[0].Sku != "test-sku" {
t.Fatalf("Unexpected SKU in GetState: %s", state2.Items[0].Sku)
}
}
// getSerializedPayload serializes a mutation proto using the registered handler.
func getSerializedPayload(handler MessageHandler, msgType uint16, content interface{}) ([]byte, error) {
msg := &Message{
Type: msgType,
Content: content,
}
var buf bytes.Buffer
if err := handler.Write(msg, &buf); err != nil {
return nil, err
}
return buf.Bytes(), nil
}
-379
View File
@@ -1,379 +0,0 @@
package main
import (
"context"
"errors"
"fmt"
"log"
"net"
"time"
proto "git.tornberg.me/go-cart-actor/proto" // underlying generated package name is 'messages'
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
// -----------------------------------------------------------------------------
// Metrics
// -----------------------------------------------------------------------------
var (
grpcMutateDuration = promauto.NewHistogram(prometheus.HistogramOpts{
Name: "cart_grpc_mutate_duration_seconds",
Help: "Duration of CartActor.Mutate RPCs",
Buckets: prometheus.DefBuckets,
})
grpcMutateErrors = promauto.NewCounter(prometheus.CounterOpts{
Name: "cart_grpc_mutate_errors_total",
Help: "Total number of failed CartActor.Mutate RPCs",
})
grpcStateDuration = promauto.NewHistogram(prometheus.HistogramOpts{
Name: "cart_grpc_get_state_duration_seconds",
Help: "Duration of CartActor.GetState RPCs",
Buckets: prometheus.DefBuckets,
})
grpcControlDuration = promauto.NewHistogram(prometheus.HistogramOpts{
Name: "cart_grpc_control_duration_seconds",
Help: "Duration of ControlPlane RPCs",
Buckets: prometheus.DefBuckets,
})
grpcControlErrors = promauto.NewCounter(prometheus.CounterOpts{
Name: "cart_grpc_control_errors_total",
Help: "Total number of failed ControlPlane RPCs",
})
)
// timeTrack wraps a closure and records duration into the supplied histogram.
func timeTrack(hist prometheus.Observer, fn func() error) (err error) {
start := time.Now()
defer func() {
hist.Observe(time.Since(start).Seconds())
}()
return fn()
}
// -----------------------------------------------------------------------------
// CartActor Service Implementation
// -----------------------------------------------------------------------------
type cartActorService struct {
proto.UnimplementedCartActorServer
pool GrainPool
}
func newCartActorService(pool GrainPool) *cartActorService {
return &cartActorService{pool: pool}
}
func (s *cartActorService) Mutate(ctx context.Context, req *proto.MutationRequest) (*proto.MutationReply, error) {
var reply *proto.MutationReply
err := timeTrack(grpcMutateDuration, func() error {
if req == nil {
return status.Error(codes.InvalidArgument, "request is nil")
}
if req.CartId == "" {
return status.Error(codes.InvalidArgument, "cart_id is empty")
}
mt := uint16(req.Type.Number())
handler, ok := Handlers[mt]
if !ok {
return status.Errorf(codes.InvalidArgument, "unknown mutation type %d", mt)
}
content, err := handler.Read(req.Payload)
if err != nil {
return status.Errorf(codes.InvalidArgument, "decode payload: %v", err)
}
ts := req.ClientTimestamp
if ts == 0 {
ts = time.Now().Unix()
}
msg := Message{
Type: mt,
TimeStamp: &ts,
Content: content,
}
frame, err := s.pool.Process(ToCartId(req.CartId), msg)
if err != nil {
return err
}
reply = &proto.MutationReply{
StatusCode: int32(frame.StatusCode),
Payload: frame.Payload,
}
return nil
})
if err != nil {
grpcMutateErrors.Inc()
return nil, err
}
return reply, nil
}
func (s *cartActorService) GetState(ctx context.Context, req *proto.StateRequest) (*proto.StateReply, error) {
var reply *proto.StateReply
err := timeTrack(grpcStateDuration, func() error {
if req == nil || req.CartId == "" {
return status.Error(codes.InvalidArgument, "cart_id is empty")
}
frame, err := s.pool.Get(ToCartId(req.CartId))
if err != nil {
return err
}
reply = &proto.StateReply{
StatusCode: int32(frame.StatusCode),
Payload: frame.Payload,
}
return nil
})
if err != nil {
return nil, err
}
return reply, nil
}
// -----------------------------------------------------------------------------
// ControlPlane Service Implementation
// -----------------------------------------------------------------------------
// controlPlaneService directly leverages SyncedPool internals (same package).
// NOTE: This is a transitional adapter; once the legacy frame-based code is
// removed, related fields/methods in SyncedPool can be slimmed.
type controlPlaneService struct {
proto.UnimplementedControlPlaneServer
pool *SyncedPool
}
func newControlPlaneService(pool *SyncedPool) *controlPlaneService {
return &controlPlaneService{pool: pool}
}
func (s *controlPlaneService) Ping(ctx context.Context, _ *proto.Empty) (*proto.PingReply, error) {
var reply *proto.PingReply
err := timeTrack(grpcControlDuration, func() error {
reply = &proto.PingReply{
Host: s.pool.Hostname,
UnixTime: time.Now().Unix(),
}
return nil
})
if err != nil {
grpcControlErrors.Inc()
return nil, err
}
return reply, nil
}
func (s *controlPlaneService) Negotiate(ctx context.Context, req *proto.NegotiateRequest) (*proto.NegotiateReply, error) {
var reply *proto.NegotiateReply
err := timeTrack(grpcControlDuration, func() error {
if req == nil {
return status.Error(codes.InvalidArgument, "request is nil")
}
// Add unknown hosts
for _, host := range req.KnownHosts {
if host == "" || host == s.pool.Hostname {
continue
}
if !s.pool.IsKnown(host) {
go s.pool.AddRemote(host)
}
}
// Build healthy host list
hosts := make([]string, 0)
for _, r := range s.pool.GetHealthyRemotes() {
hosts = append(hosts, r.Host)
}
hosts = append(hosts, s.pool.Hostname)
reply = &proto.NegotiateReply{
Hosts: hosts,
}
return nil
})
if err != nil {
grpcControlErrors.Inc()
return nil, err
}
return reply, nil
}
func (s *controlPlaneService) GetCartIds(ctx context.Context, _ *proto.Empty) (*proto.CartIdsReply, error) {
var reply *proto.CartIdsReply
err := timeTrack(grpcControlDuration, func() error {
s.pool.mu.RLock()
defer s.pool.mu.RUnlock()
ids := make([]string, 0, len(s.pool.local.grains))
for id, g := range s.pool.local.grains {
if g == nil {
continue
}
if id.String() == "" {
continue
}
ids = append(ids, id.String())
}
reply = &proto.CartIdsReply{
CartIds: ids,
}
return nil
})
if err != nil {
grpcControlErrors.Inc()
return nil, err
}
return reply, nil
}
func (s *controlPlaneService) ConfirmOwner(ctx context.Context, req *proto.OwnerChangeRequest) (*proto.OwnerChangeAck, error) {
var reply *proto.OwnerChangeAck
err := timeTrack(grpcControlDuration, func() error {
if req == nil || req.CartId == "" || req.NewHost == "" {
return status.Error(codes.InvalidArgument, "cart_id or new_host missing")
}
id := ToCartId(req.CartId)
newHost := req.NewHost
// Mirror GrainOwnerChangeHandler semantics
log.Printf("gRPC ConfirmOwner: cart %s newHost=%s", id, newHost)
for _, r := range s.pool.remoteHosts {
if r.Host == newHost && r.IsHealthy() {
go s.pool.SpawnRemoteGrain(id, newHost)
break
}
}
go s.pool.AddRemote(newHost)
reply = &proto.OwnerChangeAck{
Accepted: true,
Message: "ok",
}
return nil
})
if err != nil {
grpcControlErrors.Inc()
return nil, err
}
return reply, nil
}
func (s *controlPlaneService) Closing(ctx context.Context, notice *proto.ClosingNotice) (*proto.OwnerChangeAck, error) {
var reply *proto.OwnerChangeAck
err := timeTrack(grpcControlDuration, func() error {
if notice == nil || notice.Host == "" {
return status.Error(codes.InvalidArgument, "host missing")
}
host := notice.Host
s.pool.mu.RLock()
_, exists := s.pool.remoteHosts[host]
s.pool.mu.RUnlock()
if exists {
go s.pool.RemoveHost(host)
}
reply = &proto.OwnerChangeAck{
Accepted: true,
Message: "removed",
}
return nil
})
if err != nil {
grpcControlErrors.Inc()
return nil, err
}
return reply, nil
}
// -----------------------------------------------------------------------------
// Server Bootstrap
// -----------------------------------------------------------------------------
type GRPCServer struct {
server *grpc.Server
lis net.Listener
addr string
}
// StartGRPCServer sets up a gRPC server hosting both CartActor and ControlPlane services.
// addr example: ":1337" (for combined) OR run two servers if you want separate ports.
// For the migration we can host both on the same listener to reduce open ports.
func StartGRPCServer(addr string, pool GrainPool, synced *SyncedPool, opts ...grpc.ServerOption) (*GRPCServer, error) {
if pool == nil {
return nil, errors.New("nil grain pool")
}
if synced == nil {
return nil, errors.New("nil synced pool")
}
lis, err := net.Listen("tcp", addr)
if err != nil {
return nil, fmt.Errorf("listen %s: %w", addr, err)
}
grpcServer := grpc.NewServer(opts...)
proto.RegisterCartActorServer(grpcServer, newCartActorService(pool))
proto.RegisterControlPlaneServer(grpcServer, newControlPlaneService(synced))
go func() {
log.Printf("gRPC server listening on %s", addr)
if serveErr := grpcServer.Serve(lis); serveErr != nil {
log.Printf("gRPC server stopped: %v", serveErr)
}
}()
return &GRPCServer{
server: grpcServer,
lis: lis,
addr: addr,
}, nil
}
// GracefulStop stops the server gracefully.
func (s *GRPCServer) GracefulStop() {
if s == nil || s.server == nil {
return
}
s.server.GracefulStop()
}
// Addr returns the bound address.
func (s *GRPCServer) Addr() string {
if s == nil {
return ""
}
return s.addr
}
// -----------------------------------------------------------------------------
// Client Dial Helpers (used later by refactored remote grain + control plane)
// -----------------------------------------------------------------------------
// DialRemote establishes (or reuses externally) a gRPC client connection.
func DialRemote(ctx context.Context, target string, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
dialOpts := []grpc.DialOption{
grpc.WithInsecure(), // NOTE: Intentional for initial migration; replace with TLS / mTLS later.
grpc.WithBlock(),
}
dialOpts = append(dialOpts, opts...)
ctxDial, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
conn, err := grpc.DialContext(ctxDial, target, dialOpts...)
if err != nil {
return nil, err
}
return conn, nil
}
// -----------------------------------------------------------------------------
// Utility for converting internal errors to gRPC status (if needed later).
// -----------------------------------------------------------------------------
func grpcError(err error) error {
if err == nil {
return nil
}
// Extend mapping if we add richer error types.
return status.Error(codes.Internal, err.Error())
}
-128
View File
@@ -1,128 +0,0 @@
package main
import (
"encoding/json"
"fmt"
"io"
"log"
"net/http"
"github.com/google/uuid"
)
type KlarnaClient struct {
Url string
UserName string
Password string
client *http.Client
}
func NewKlarnaClient(url, userName, password string) *KlarnaClient {
return &KlarnaClient{
Url: url,
UserName: userName,
Password: password,
client: &http.Client{},
}
}
const (
KlarnaPlaygroundUrl = "https://api.playground.klarna.com"
)
func (k *KlarnaClient) GetOrder(orderId string) (*CheckoutOrder, error) {
req, err := http.NewRequest("GET", k.Url+"/checkout/v3/orders/"+orderId, nil)
if err != nil {
return nil, err
}
req.Header.Add("Content-Type", "application/json")
req.SetBasicAuth(k.UserName, k.Password)
res, err := k.client.Do(req)
if err != nil {
return nil, err
}
defer res.Body.Close()
return k.getOrderResponse(res)
}
func (k *KlarnaClient) getOrderResponse(res *http.Response) (*CheckoutOrder, error) {
var err error
var klarnaOrderResponse CheckoutOrder
if res.StatusCode >= 200 && res.StatusCode <= 299 {
err = json.NewDecoder(res.Body).Decode(&klarnaOrderResponse)
if err != nil {
return nil, err
}
return &klarnaOrderResponse, nil
}
body, err := io.ReadAll(res.Body)
if err == nil {
log.Println(string(body))
}
return nil, fmt.Errorf("%s", res.Status)
}
func (k *KlarnaClient) CreateOrder(reader io.Reader) (*CheckoutOrder, error) {
//bytes.NewReader(reply.Payload)
req, err := http.NewRequest("POST", k.Url+"/checkout/v3/orders", reader)
if err != nil {
return nil, err
}
req.Header.Add("Content-Type", "application/json")
req.SetBasicAuth(k.UserName, k.Password)
res, err := http.DefaultClient.Do(req)
if nil != err {
return nil, err
}
defer res.Body.Close()
return k.getOrderResponse(res)
}
func (k *KlarnaClient) UpdateOrder(orderId string, reader io.Reader) (*CheckoutOrder, error) {
//bytes.NewReader(reply.Payload)
req, err := http.NewRequest("POST", fmt.Sprintf("%s/checkout/v3/orders/%s", k.Url, orderId), reader)
if err != nil {
return nil, err
}
req.Header.Add("Content-Type", "application/json")
req.SetBasicAuth(k.UserName, k.Password)
res, err := http.DefaultClient.Do(req)
if nil != err {
return nil, err
}
defer res.Body.Close()
return k.getOrderResponse(res)
}
func (k *KlarnaClient) AbortOrder(orderId string) error {
req, err := http.NewRequest("POST", fmt.Sprintf("%s/checkout/v3/orders/%s/abort", k.Url, orderId), nil)
if err != nil {
return err
}
req.SetBasicAuth(k.UserName, k.Password)
_, err = http.DefaultClient.Do(req)
return err
}
// ordermanagement/v1/orders/{order_id}/acknowledge
func (k *KlarnaClient) AcknowledgeOrder(orderId string) error {
req, err := http.NewRequest("POST", fmt.Sprintf("%s/ordermanagement/v1/orders/%s/acknowledge", k.Url, orderId), nil)
if err != nil {
return err
}
id := uuid.New()
req.SetBasicAuth(k.UserName, k.Password)
req.Header.Add("Klarna-Idempotency-Key", id.String())
_, err = http.DefaultClient.Do(req)
return err
}
-169
View File
@@ -1,169 +0,0 @@
package main
type (
LineType string
// CheckoutOrder type is the request structure to create a new order from the Checkout API
CheckoutOrder struct {
ID string `json:"order_id,omitempty"`
PurchaseCountry string `json:"purchase_country"`
PurchaseCurrency string `json:"purchase_currency"`
Locale string `json:"locale"`
Status string `json:"status,omitempty"`
BillingAddress *Address `json:"billing_address,omitempty"`
ShippingAddress *Address `json:"shipping_address,omitempty"`
OrderAmount int `json:"order_amount"`
OrderTaxAmount int `json:"order_tax_amount"`
OrderLines []*Line `json:"order_lines"`
Customer *CheckoutCustomer `json:"customer,omitempty"`
MerchantURLS *CheckoutMerchantURLS `json:"merchant_urls"`
HTMLSnippet string `json:"html_snippet,omitempty"`
MerchantReference1 string `json:"merchant_reference1,omitempty"`
MerchantReference2 string `json:"merchant_reference2,omitempty"`
StartedAt string `json:"started_at,omitempty"`
CompletedAt string `json:"completed_at,omitempty"`
LastModifiedAt string `json:"last_modified_at,omitempty"`
Options *CheckoutOptions `json:"options,omitempty"`
Attachment *Attachment `json:"attachment,omitempty"`
ExternalPaymentMethods []*PaymentProvider `json:"external_payment_methods,omitempty"`
ExternalCheckouts []*PaymentProvider `json:"external_checkouts,omitempty"`
ShippingCountries []string `json:"shipping_countries,omitempty"`
ShippingOptions []*ShippingOption `json:"shipping_options,omitempty"`
MerchantData string `json:"merchant_data,omitempty"`
GUI *GUI `json:"gui,omitempty"`
MerchantRequested *AdditionalCheckBox `json:"merchant_requested,omitempty"`
SelectedShippingOption *ShippingOption `json:"selected_shipping_option,omitempty"`
ErrorCode string `json:"error_code,omitempty"`
ErrorMessages []string `json:"error_messages,omitempty"`
}
// GUI type wraps the GUI options
GUI struct {
Options []string `json:"options,omitempty"`
}
// ShippingOption type is part of the CheckoutOrder structure, represent the shipping options field
ShippingOption struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description,omitempty"`
Promo string `json:"promo,omitempty"`
Price int `json:"price"`
TaxAmount int `json:"tax_amount"`
TaxRate int `json:"tax_rate"`
Preselected bool `json:"preselected,omitempty"`
ShippingMethod string `json:"shipping_method,omitempty"`
}
// PaymentProvider type is part of the CheckoutOrder structure, represent the ExternalPaymentMethods and
// ExternalCheckouts field
PaymentProvider struct {
Name string `json:"name"`
RedirectURL string `json:"redirect_url"`
ImageURL string `json:"image_url,omitempty"`
Fee int `json:"fee,omitempty"`
Description string `json:"description,omitempty"`
Countries []string `json:"countries,omitempty"`
}
Attachment struct {
ContentType string `json:"content_type"`
Body string `json:"body"`
}
CheckoutOptions struct {
AcquiringChannel string `json:"acquiring_channel,omitempty"`
AllowSeparateShippingAddress bool `json:"allow_separate_shipping_address,omitempty"`
ColorButton string `json:"color_button,omitempty"`
ColorButtonText string `json:"color_button_text,omitempty"`
ColorCheckbox string `json:"color_checkbox,omitempty"`
ColorCheckboxCheckmark string `json:"color_checkbox_checkmark,omitempty"`
ColorHeader string `json:"color_header,omitempty"`
ColorLink string `json:"color_link,omitempty"`
DateOfBirthMandatory bool `json:"date_of_birth_mandatory,omitempty"`
ShippingDetails string `json:"shipping_details,omitempty"`
TitleMandatory bool `json:"title_mandatory,omitempty"`
AdditionalCheckbox *AdditionalCheckBox `json:"additional_checkbox"`
RadiusBorder string `json:"radius_border,omitempty"`
ShowSubtotalDetail bool `json:"show_subtotal_detail,omitempty"`
RequireValidateCallbackSuccess bool `json:"require_validate_callback_success,omitempty"`
AllowGlobalBillingCountries bool `json:"allow_global_billing_countries,omitempty"`
}
AdditionalCheckBox struct {
Text string `json:"text"`
Checked bool `json:"checked"`
Required bool `json:"required"`
}
CheckoutMerchantURLS struct {
// URL of merchant terms and conditions. Should be different than checkout, confirmation and push URLs.
// (max 2000 characters)
Terms string `json:"terms"`
// URL of merchant checkout page. Should be different than terms, confirmation and push URLs.
// (max 2000 characters)
Checkout string `json:"checkout"`
// URL of merchant confirmation page. Should be different than checkout and confirmation URLs.
// (max 2000 characters)
Confirmation string `json:"confirmation"`
// URL that will be requested when an order is completed. Should be different than checkout and
// confirmation URLs. (max 2000 characters)
Push string `json:"push"`
// URL that will be requested for final merchant validation. (must be https, max 2000 characters)
Validation string `json:"validation,omitempty"`
// URL for shipping option update. (must be https, max 2000 characters)
ShippingOptionUpdate string `json:"shipping_option_update,omitempty"`
// URL for shipping, tax and purchase currency updates. Will be called on address changes.
// (must be https, max 2000 characters)
AddressUpdate string `json:"address_update,omitempty"`
// URL for notifications on pending orders. (max 2000 characters)
Notification string `json:"notification,omitempty"`
// URL for shipping, tax and purchase currency updates. Will be called on purchase country changes.
// (must be https, max 2000 characters)
CountryChange string `json:"country_change,omitempty"`
}
CheckoutCustomer struct {
// DateOfBirth in string representation 2006-01-02
DateOfBirth string `json:"date_of_birth"`
}
// Address type define the address object (json serializable) being used for the API to represent billing &
// shipping addresses
Address struct {
GivenName string `json:"given_name,omitempty"`
FamilyName string `json:"family_name,omitempty"`
Email string `json:"email,omitempty"`
Title string `json:"title,omitempty"`
StreetAddress string `json:"street_address,omitempty"`
StreetAddress2 string `json:"street_address2,omitempty"`
PostalCode string `json:"postal_code,omitempty"`
City string `json:"city,omitempty"`
Region string `json:"region,omitempty"`
Phone string `json:"phone,omitempty"`
Country string `json:"country,omitempty"`
}
Line struct {
Type string `json:"type,omitempty"`
Reference string `json:"reference,omitempty"`
Name string `json:"name"`
Quantity int `json:"quantity"`
QuantityUnit string `json:"quantity_unit,omitempty"`
UnitPrice int `json:"unit_price"`
TaxRate int `json:"tax_rate"`
TotalAmount int `json:"total_amount"`
TotalDiscountAmount int `json:"total_discount_amount,omitempty"`
TotalTaxAmount int `json:"total_tax_amount"`
MerchantData string `json:"merchant_data,omitempty"`
ProductURL string `json:"product_url,omitempty"`
ImageURL string `json:"image_url,omitempty"`
}
)
+62 -346
View File
@@ -2,47 +2,18 @@ package main
import ( import (
"encoding/json" "encoding/json"
"fmt"
"log" "log"
"net/http" "net/http"
"net/http/pprof"
"os" "os"
"os/signal"
"strings"
"syscall"
"time" "time"
messages "git.tornberg.me/go-cart-actor/proto" messages "git.tornberg.me/go-cart-actor/proto"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"github.com/prometheus/client_golang/prometheus/promhttp"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
)
var (
grainSpawns = promauto.NewCounter(prometheus.CounterOpts{
Name: "cart_grain_spawned_total",
Help: "The total number of spawned grains",
})
grainMutations = promauto.NewCounter(prometheus.CounterOpts{
Name: "cart_grain_mutations_total",
Help: "The total number of mutations",
})
grainLookups = promauto.NewCounter(prometheus.CounterOpts{
Name: "cart_grain_lookups_total",
Help: "The total number of lookups",
})
) )
func spawn(id CartId) (*CartGrain, error) { func spawn(id CartId) (*CartGrain, error) {
grainSpawns.Inc()
ret := &CartGrain{ ret := &CartGrain{
lastItemId: 0,
lastDeliveryId: 0,
Deliveries: []*CartDelivery{},
Id: id, Id: id,
Items: []*CartItem{}, Items: []CartItem{},
storageMessages: []Message{}, storageMessages: []Message{},
TotalPrice: 0, TotalPrice: 0,
} }
@@ -59,25 +30,42 @@ type App struct {
storage *DiskStorage storage *DiskStorage
} }
func (a *App) Save() error { func (a *App) HandleGet(w http.ResponseWriter, r *http.Request) {
hasChanges := false id := r.PathValue("id")
a.pool.mu.RLock() grain, err := a.pool.Get(ToCartId(id))
defer a.pool.mu.RUnlock() if err != nil {
for id, grain := range a.pool.GetGrains() { w.WriteHeader(http.StatusNotFound)
if grain == nil { return
continue }
} w.Header().Set("Content-Type", "application/json")
if grain.GetLastChange() > a.storage.LastSaves[id] { w.WriteHeader(http.StatusOK)
hasChanges = true json.NewEncoder(w).Encode(grain)
err := a.storage.Store(id, grain) }
if err != nil {
log.Printf("Error saving grain %s: %v\n", id, err) func (a *App) HandleAddSku(w http.ResponseWriter, r *http.Request) {
} id := r.PathValue("id")
} sku := r.PathValue("sku")
grain, err := a.pool.Process(ToCartId(id), Message{
Type: AddRequestType,
Content: &messages.AddRequest{Sku: sku},
})
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error()))
return
} }
if !hasChanges { w.Header().Set("Content-Type", "application/json")
return nil w.WriteHeader(http.StatusOK)
json.NewEncoder(w).Encode(grain)
}
func (a *App) Save() error {
for id, grain := range a.pool.GetGrains() {
err := a.storage.Store(id, grain)
if err != nil {
log.Printf("Error saving grain %s: %v\n", id, err)
}
} }
return a.storage.saveState() return a.storage.saveState()
} }
@@ -92,330 +80,58 @@ func (a *App) HandleSave(w http.ResponseWriter, r *http.Request) {
} }
} }
var podIp = os.Getenv("POD_IP")
var name = os.Getenv("POD_NAME")
var amqpUrl = os.Getenv("AMQP_URL")
var KlarnaInstance = NewKlarnaClient(KlarnaPlaygroundUrl, os.Getenv("KLARNA_API_USERNAME"), os.Getenv("KLARNA_API_PASSWORD"))
func GetDiscovery() Discovery {
if podIp == "" {
return nil
}
config, kerr := rest.InClusterConfig()
if kerr != nil {
log.Fatalf("Error creating kubernetes client: %v\n", kerr)
}
client, err := kubernetes.NewForConfig(config)
if err != nil {
log.Fatalf("Error creating client: %v\n", err)
}
return NewK8sDiscovery(client)
}
var tpl = `<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>s10r testing - checkout</title>
</head>
<body>
%s
</body>
</html>
`
func getCountryFromHost(host string) string {
if strings.Contains(strings.ToLower(host), "-no") {
return "no"
}
return "se"
}
func getCheckoutOrder(host string, cartId CartId) *messages.CreateCheckoutOrder {
baseUrl := fmt.Sprintf("https://%s", host)
cartBaseUrl := os.Getenv("CART_BASE_URL")
if cartBaseUrl == "" {
cartBaseUrl = "https://cart.tornberg.me"
}
country := getCountryFromHost(host)
return &messages.CreateCheckoutOrder{
Terms: fmt.Sprintf("%s/terms", baseUrl),
Checkout: fmt.Sprintf("%s/checkout?order_id={checkout.order.id}", baseUrl),
Confirmation: fmt.Sprintf("%s/confirmation/{checkout.order.id}", baseUrl),
Validation: fmt.Sprintf("%s/validation", cartBaseUrl),
Push: fmt.Sprintf("%s/push?order_id={checkout.order.id}", cartBaseUrl),
Country: country,
}
}
func main() { func main() {
// Create a new instance of the server
storage, err := NewDiskStorage(fmt.Sprintf("data/%s_state.gob", name)) storage, err := NewDiskStorage("data/state.gob")
if err != nil { if err != nil {
log.Printf("Error loading state: %v\n", err) log.Printf("Error loading state: %v\n", err)
} }
app := &App{ app := &App{
pool: NewGrainLocalPool(65535, 5*time.Minute, spawn), pool: NewGrainLocalPool(1000, 5*time.Minute, spawn),
storage: storage, storage: storage,
} }
syncedPool, err := NewSyncedPool(app.pool, podIp, GetDiscovery()) rpcHandler, err := NewGrainHandler(app.pool, "localhost:1337")
if err != nil { if err != nil {
log.Fatalf("Error creating synced pool: %v\n", err) log.Fatalf("Error creating handler: %v\n", err)
} }
go rpcHandler.Serve()
// Start unified gRPC server (CartActor + ControlPlane) replacing legacy RPC server on :1337 remotePool := NewRemoteGrainPool("localhost:1337")
// TODO: Remove any remaining legacy RPC server references and deprecated frame-based code after full gRPC migration is validated.
grpcSrv, err := StartGRPCServer(":1337", app.pool, syncedPool)
if err != nil {
log.Fatalf("Error starting gRPC server: %v\n", err)
}
defer grpcSrv.GracefulStop()
go func() {
for range time.Tick(time.Minute * 10) {
err := app.Save()
if err != nil {
log.Printf("Error saving: %v\n", err)
}
}
}()
orderHandler := &AmqpOrderHandler{
Url: amqpUrl,
}
syncedServer := NewPoolServer(syncedPool, fmt.Sprintf("%s, %s", name, podIp))
mux := http.NewServeMux() mux := http.NewServeMux()
mux.Handle("/cart/", http.StripPrefix("/cart", syncedServer.Serve())) mux.HandleFunc("GET /api/{id}", app.HandleGet)
// only for local mux.HandleFunc("GET /api/{id}/add/{sku}", app.HandleAddSku)
// mux.HandleFunc("GET /add/remote/{host}", func(w http.ResponseWriter, r *http.Request) { mux.HandleFunc("GET /remote/{id}/add", func(w http.ResponseWriter, r *http.Request) {
// syncedPool.AddRemote(r.PathValue("host")) id := r.PathValue("id")
// }) ts := time.Now().Unix()
// mux.HandleFunc("GET /save", app.HandleSave) data, err := remotePool.Process(ToCartId(id), Message{
//mux.HandleFunc("/", app.RewritePath) Type: AddRequestType,
mux.HandleFunc("/debug/pprof/", pprof.Index) TimeStamp: &ts,
mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline) Content: &messages.AddRequest{Sku: "49565"},
mux.HandleFunc("/debug/pprof/profile", pprof.Profile) })
mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
mux.Handle("/metrics", promhttp.Handler())
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
// Grain pool health: simple capacity check (mirrors previous GrainHandler.IsHealthy)
app.pool.mu.RLock()
grainCount := len(app.pool.grains)
capacity := app.pool.PoolSize
app.pool.mu.RUnlock()
if grainCount >= capacity {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte("grain pool at capacity"))
return
}
if !syncedPool.IsHealthy() {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte("control plane not healthy"))
return
}
w.WriteHeader(http.StatusOK)
w.Write([]byte("ok"))
})
mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte("ok"))
})
mux.HandleFunc("/livez", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
w.Write([]byte("ok"))
})
mux.HandleFunc("/checkout", func(w http.ResponseWriter, r *http.Request) {
orderId := r.URL.Query().Get("order_id")
order := &CheckoutOrder{}
if orderId == "" {
cookie, err := r.Cookie("cartid")
if err != nil {
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte(err.Error()))
return
}
if cookie.Value == "" {
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte("no cart id to checkout is empty"))
return
}
cartId := ToCartId(cookie.Value)
reply, err := syncedServer.pool.Process(cartId, Message{
Type: CreateCheckoutOrderType,
Content: getCheckoutOrder(r.Host, cartId),
})
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error()))
}
err = json.Unmarshal(reply.Payload, &order)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error()))
return
}
} else {
prevOrder, err := KlarnaInstance.GetOrder(orderId)
if err != nil {
w.WriteHeader(http.StatusBadRequest)
w.Write([]byte(err.Error()))
return
}
order = prevOrder
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
w.Header().Set("Permissions-Policy", "payment=(self \"https://js.stripe.com\" \"https://m.stripe.network\" \"https://js.playground.kustom.co\")")
w.WriteHeader(http.StatusOK)
w.Write([]byte(fmt.Sprintf(tpl, order.HTMLSnippet)))
})
mux.HandleFunc("/confirmation/{order_id}", func(w http.ResponseWriter, r *http.Request) {
orderId := r.PathValue("order_id")
order, err := KlarnaInstance.GetOrder(orderId)
if err != nil { if err != nil {
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error())) w.Write([]byte(err.Error()))
return return
} }
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Content-Type", "text/html; charset=utf-8")
if order.Status == "checkout_complete" {
http.SetCookie(w, &http.Cookie{
Name: "cartid",
Value: "",
Path: "/",
Secure: true,
HttpOnly: true,
Expires: time.Unix(0, 0),
SameSite: http.SameSiteLaxMode,
})
}
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
w.Write([]byte(fmt.Sprintf(tpl, order.HTMLSnippet))) w.Write(data)
}) })
mux.HandleFunc("/validate", func(w http.ResponseWriter, r *http.Request) { mux.HandleFunc("GET /remote/{id}", func(w http.ResponseWriter, r *http.Request) {
log.Printf("Klarna order validation, method: %s", r.Method) id := r.PathValue("id")
if r.Method != "POST" { data, err := remotePool.Get(ToCartId(id))
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
order := &CheckoutOrder{}
err := json.NewDecoder(r.Body).Decode(order)
if err != nil { if err != nil {
w.WriteHeader(http.StatusBadRequest)
}
log.Printf("Klarna order validation: %s", order.ID)
//err = confirmOrder(order, orderHandler)
//if err != nil {
// log.Printf("Error validating order: %v\n", err)
// w.WriteHeader(http.StatusInternalServerError)
// return
//}
//
//err = triggerOrderCompleted(err, syncedServer, order)
//if err != nil {
// log.Printf("Error processing cart message: %v\n", err)
// w.WriteHeader(http.StatusInternalServerError)
// return
//}
w.WriteHeader(http.StatusOK)
})
mux.HandleFunc("/push", func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
orderId := r.URL.Query().Get("order_id")
log.Printf("Order confirmation push: %s", orderId)
order, err := KlarnaInstance.GetOrder(orderId)
if err != nil {
log.Printf("Error creating request: %v\n", err)
w.WriteHeader(http.StatusInternalServerError) w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error()))
return return
} }
w.Header().Set("Content-Type", "application/json")
err = confirmOrder(order, orderHandler)
if err != nil {
log.Printf("Error confirming order: %v\n", err)
w.WriteHeader(http.StatusInternalServerError)
return
}
err = triggerOrderCompleted(err, syncedServer, order)
if err != nil {
log.Printf("Error processing cart message: %v\n", err)
w.WriteHeader(http.StatusInternalServerError)
return
}
err = KlarnaInstance.AcknowledgeOrder(orderId)
if err != nil {
log.Printf("Error acknowledging order: %v\n", err)
}
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
w.Write(data)
}) })
mux.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) { mux.HandleFunc("GET /save", app.HandleSave)
w.WriteHeader(http.StatusOK) http.ListenAndServe(":8080", mux)
w.Write([]byte("1.0.0"))
})
sigs := make(chan os.Signal, 1)
done := make(chan bool, 1)
signal.Notify(sigs, syscall.SIGTERM)
go func() {
sig := <-sigs
fmt.Println("Shutting down due to signal:", sig)
go syncedPool.Close()
app.Save()
done <- true
}()
log.Print("Server started at port 8080")
go http.ListenAndServe(":8080", mux)
<-done
} }
func triggerOrderCompleted(err error, syncedServer *PoolServer, order *CheckoutOrder) error {
_, err = syncedServer.pool.Process(ToCartId(order.MerchantReference1), Message{
Type: OrderCompletedType,
Content: &messages.OrderCreated{
OrderId: order.ID,
Status: order.Status,
},
})
return err
}
func confirmOrder(order *CheckoutOrder, orderHandler *AmqpOrderHandler) error {
orderToSend, err := json.Marshal(order)
if err != nil {
return err
}
err = orderHandler.Connect()
if err != nil {
return err
}
defer orderHandler.Close()
err = orderHandler.OrderCompleted(orderToSend)
if err != nil {
return err
}
return nil
}
-315
View File
@@ -1,315 +0,0 @@
package main
import (
"fmt"
"io"
messages "git.tornberg.me/go-cart-actor/proto"
"google.golang.org/protobuf/proto"
)
var Handlers = map[uint16]MessageHandler{
AddRequestType: &AddRequestHandler{},
AddItemType: &AddItemHandler{},
ChangeQuantityType: &ChangeQuantityHandler{},
SetDeliveryType: &SetDeliveryHandler{},
RemoveItemType: &RemoveItemHandler{},
RemoveDeliveryType: &RemoveDeliveryHandler{},
CreateCheckoutOrderType: &CheckoutHandler{},
SetCartItemsType: &SetCartItemsHandler{},
OrderCompletedType: &OrderCompletedHandler{},
}
func GetMessageHandler(t uint16) (MessageHandler, error) {
h, ok := Handlers[t]
if !ok {
return nil, fmt.Errorf("no handler for message type %d", t)
}
return h, nil
}
type MessageHandler interface {
Write(*Message, io.Writer) error
Read(data []byte) (interface{}, error)
Is(*Message) bool
}
type TypedMessageHandler struct {
Type uint16
}
type SetCartItemsHandler struct {
TypedMessageHandler
}
func (h *SetCartItemsHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.SetCartRequest))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *SetCartItemsHandler) Read(data []byte) (interface{}, error) {
msg := &messages.SetCartRequest{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *SetCartItemsHandler) Is(m *Message) bool {
if m.Type != AddRequestType {
return false
}
_, ok := m.Content.(*messages.SetCartRequest)
return ok
}
type AddRequestHandler struct {
TypedMessageHandler
}
func (h *AddRequestHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.AddRequest))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *AddRequestHandler) Read(data []byte) (interface{}, error) {
msg := &messages.AddRequest{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *AddRequestHandler) Is(m *Message) bool {
if m.Type != AddRequestType {
return false
}
_, ok := m.Content.(*messages.AddRequest)
return ok
}
type AddItemHandler struct {
TypedMessageHandler
}
func (h *AddItemHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.AddItem))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *AddItemHandler) Read(data []byte) (interface{}, error) {
msg := &messages.AddItem{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *AddItemHandler) Is(m *Message) bool {
if m.Type != AddItemType {
return false
}
_, ok := m.Content.(*messages.AddItem)
return ok
}
type ChangeQuantityHandler struct {
TypedMessageHandler
}
func (h *ChangeQuantityHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.ChangeQuantity))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *ChangeQuantityHandler) Read(data []byte) (interface{}, error) {
msg := &messages.ChangeQuantity{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *ChangeQuantityHandler) Is(m *Message) bool {
if m.Type != ChangeQuantityType {
return false
}
_, ok := m.Content.(*messages.ChangeQuantity)
return ok
}
type SetDeliveryHandler struct {
TypedMessageHandler
}
func (h *SetDeliveryHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.SetDelivery))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *SetDeliveryHandler) Read(data []byte) (interface{}, error) {
msg := &messages.SetDelivery{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *SetDeliveryHandler) Is(m *Message) bool {
if m.Type != ChangeQuantityType {
return false
}
_, ok := m.Content.(*messages.SetDelivery)
return ok
}
type RemoveItemHandler struct {
TypedMessageHandler
}
func (h *RemoveItemHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.RemoveItem))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *RemoveItemHandler) Read(data []byte) (interface{}, error) {
msg := &messages.RemoveItem{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *RemoveItemHandler) Is(m *Message) bool {
if m.Type != AddItemType {
return false
}
_, ok := m.Content.(*messages.RemoveItem)
return ok
}
type RemoveDeliveryHandler struct {
TypedMessageHandler
}
func (h *RemoveDeliveryHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.RemoveDelivery))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *RemoveDeliveryHandler) Read(data []byte) (interface{}, error) {
msg := &messages.RemoveDelivery{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *RemoveDeliveryHandler) Is(m *Message) bool {
if m.Type != AddItemType {
return false
}
_, ok := m.Content.(*messages.RemoveDelivery)
return ok
}
type CheckoutHandler struct {
TypedMessageHandler
}
func (h *CheckoutHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.CreateCheckoutOrder))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *CheckoutHandler) Read(data []byte) (interface{}, error) {
msg := &messages.CreateCheckoutOrder{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *CheckoutHandler) Is(m *Message) bool {
if m.Type != CreateCheckoutOrderType {
return false
}
_, ok := m.Content.(*messages.CreateCheckoutOrder)
return ok
}
type OrderCompletedHandler struct {
TypedMessageHandler
}
func (h *OrderCompletedHandler) Write(m *Message, w io.Writer) error {
messageBytes, err := proto.Marshal(m.Content.(*messages.OrderCreated))
if err != nil {
return err
}
w.Write(messageBytes)
return nil
}
func (h *OrderCompletedHandler) Read(data []byte) (interface{}, error) {
msg := &messages.OrderCreated{}
err := proto.Unmarshal(data, msg)
if err != nil {
return nil, err
}
return msg, nil
}
func (h *OrderCompletedHandler) Is(m *Message) bool {
if m.Type != OrderCompletedType {
return false
}
_, ok := m.Content.(*messages.OrderCreated)
return ok
}
-9
View File
@@ -3,13 +3,4 @@ package main
const ( const (
AddRequestType = 1 AddRequestType = 1
AddItemType = 2 AddItemType = 2
RemoveItemType = 4
RemoveDeliveryType = 5
ChangeQuantityType = 6
SetDeliveryType = 7
SetPickupPointType = 8
CreateCheckoutOrderType = 9
SetCartItemsType = 10
OrderCompletedType = 11
) )
+32 -15
View File
@@ -3,8 +3,12 @@ package main
import ( import (
"bytes" "bytes"
"encoding/binary" "encoding/binary"
"fmt"
"io" "io"
"time" "time"
messages "git.tornberg.me/go-cart-actor/proto"
"google.golang.org/protobuf/proto"
) )
type StorableMessage interface { type StorableMessage interface {
@@ -39,12 +43,21 @@ func GetData(fn func(w io.Writer) error) ([]byte, error) {
} }
func (m Message) Write(w io.Writer) error { func (m Message) Write(w io.Writer) error {
h, err := GetMessageHandler(m.Type) data, err := GetData(func(wr io.Writer) error {
if err != nil { if m.Type == AddRequestType {
return err messageBytes, err := proto.Marshal(m.Content.(*messages.AddRequest))
} if err != nil {
data, err := GetData(func(w io.Writer) error { return err
return h.Write(&m, w) }
wr.Write(messageBytes)
} else if m.Type == AddItemType {
messageBytes, err := proto.Marshal(m.Content.(*messages.AddItem))
if err != nil {
return err
}
wr.Write(messageBytes)
}
return nil
}) })
if err != nil { if err != nil {
return err return err
@@ -65,8 +78,7 @@ func (m Message) Write(w io.Writer) error {
return err return err
} }
func ReadMessage(reader io.Reader, m *Message) error { func MessageFromReader(reader io.Reader, m *Message) error {
header := StorableMessageHeader{} header := StorableMessageHeader{}
err := binary.Read(reader, binary.LittleEndian, &header) err := binary.Read(reader, binary.LittleEndian, &header)
if err != nil { if err != nil {
@@ -77,16 +89,21 @@ func ReadMessage(reader io.Reader, m *Message) error {
if err != nil { if err != nil {
return err return err
} }
h, err := GetMessageHandler(header.Type) switch header.Type {
case AddRequestType:
msg := &messages.AddRequest{}
err = proto.Unmarshal(messageBytes, msg)
m.Content = msg
case AddItemType:
msg := &messages.AddItem{}
err = proto.Unmarshal(messageBytes, msg)
m.Content = msg
default:
return fmt.Errorf("unknown message type")
}
if err != nil { if err != nil {
return err return err
} }
content, err := h.Read(messageBytes)
if err != nil {
return err
}
m.Content = content
m.Type = header.Type m.Type = header.Type
m.TimeStamp = &header.TimeStamp m.TimeStamp = &header.TimeStamp
+76
View File
@@ -0,0 +1,76 @@
package main
import (
"encoding/binary"
"io"
)
const (
RemoteGetState = uint16(0x01)
RemoteHandleMessage = uint16(0x02)
ResponseBody = uint16(0x03)
)
type CartPacket struct {
Version uint16
MessageType uint16
Id CartId
DataLength uint16
}
type ResponsePacket struct {
Version uint16
MessageType uint16
DataLength uint16
}
func SendCartPacket(conn io.Writer, id CartId, messageType uint16, datafn func(w io.Writer) error) error {
data, err := GetData(datafn)
if err != nil {
return err
}
binary.Write(conn, binary.LittleEndian, CartPacket{
Version: 2,
MessageType: messageType,
Id: id,
DataLength: uint16(len(data)),
})
_, err = conn.Write(data)
return err
}
func SendPacket(conn io.Writer, messageType uint16, datafn func(w io.Writer) error) error {
data, err := GetData(datafn)
if err != nil {
return err
}
binary.Write(conn, binary.LittleEndian, ResponsePacket{
Version: 1,
MessageType: messageType,
DataLength: uint16(len(data)),
})
_, err = conn.Write(data)
return err
}
// func ReceiveCartPacket(conn io.Reader) (CartPacket, []byte, error) {
// var packet CartPacket
// err := binary.Read(conn, binary.LittleEndian, &packet)
// if err != nil {
// return packet, nil, err
// }
// data := make([]byte, packet.DataLength)
// _, err = conn.Read(data)
// return packet, data, err
// }
func ReceivePacket(conn io.Reader) (uint16, []byte, error) {
var packet ResponsePacket
err := binary.Read(conn, binary.LittleEndian, &packet)
if err != nil {
return packet.MessageType, nil, err
}
data := make([]byte, packet.DataLength)
_, err = conn.Read(data)
return packet.MessageType, data, err
}
-349
View File
@@ -1,349 +0,0 @@
package main
import (
"encoding/json"
"fmt"
"log"
"math/rand"
"net/http"
"strconv"
"time"
messages "git.tornberg.me/go-cart-actor/proto"
)
type PoolServer struct {
pod_name string
pool GrainPool
}
func NewPoolServer(pool GrainPool, pod_name string) *PoolServer {
return &PoolServer{
pod_name: pod_name,
pool: pool,
}
}
func (s *PoolServer) HandleGet(w http.ResponseWriter, r *http.Request, id CartId) error {
data, err := s.pool.Get(id)
if err != nil {
return err
}
return s.WriteResult(w, data)
}
func (s *PoolServer) HandleAddSku(w http.ResponseWriter, r *http.Request, id CartId) error {
sku := r.PathValue("sku")
data, err := s.pool.Process(id, Message{
Type: AddRequestType,
Content: &messages.AddRequest{Sku: sku, Quantity: 1},
})
if err != nil {
return err
}
return s.WriteResult(w, data)
}
func ErrorHandler(fn func(w http.ResponseWriter, r *http.Request) error) func(w http.ResponseWriter, r *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
err := fn(w, r)
if err != nil {
log.Printf("Server error, not remote error: %v\n", err)
w.WriteHeader(http.StatusInternalServerError)
w.Write([]byte(err.Error()))
}
}
}
func (s *PoolServer) WriteResult(w http.ResponseWriter, result *FrameWithPayload) error {
w.Header().Set("Content-Type", "application/json")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("X-Pod-Name", s.pod_name)
if result.StatusCode != 200 {
log.Printf("Call error: %d\n", result.StatusCode)
if result.StatusCode >= 200 && result.StatusCode < 600 {
w.WriteHeader(int(result.StatusCode))
} else {
w.WriteHeader(http.StatusInternalServerError)
}
w.Write([]byte(result.Payload))
return nil
}
w.WriteHeader(http.StatusOK)
_, err := w.Write(result.Payload)
return err
}
func (s *PoolServer) HandleDeleteItem(w http.ResponseWriter, r *http.Request, id CartId) error {
itemIdString := r.PathValue("itemId")
itemId, err := strconv.Atoi(itemIdString)
if err != nil {
return err
}
data, err := s.pool.Process(id, Message{
Type: RemoveItemType,
Content: &messages.RemoveItem{Id: int64(itemId)},
})
if err != nil {
return err
}
return s.WriteResult(w, data)
}
type SetDelivery struct {
Provider string `json:"provider"`
Items []int64 `json:"items"`
PickupPoint *messages.PickupPoint `json:"pickupPoint,omitempty"`
}
func (s *PoolServer) HandleSetDelivery(w http.ResponseWriter, r *http.Request, id CartId) error {
delivery := SetDelivery{}
err := json.NewDecoder(r.Body).Decode(&delivery)
if err != nil {
return err
}
data, err := s.pool.Process(id, Message{
Type: SetDeliveryType,
Content: &messages.SetDelivery{
Provider: delivery.Provider,
Items: delivery.Items,
PickupPoint: delivery.PickupPoint,
},
})
if err != nil {
return err
}
return s.WriteResult(w, data)
}
func (s *PoolServer) HandleSetPickupPoint(w http.ResponseWriter, r *http.Request, id CartId) error {
deliveryIdString := r.PathValue("deliveryId")
deliveryId, err := strconv.Atoi(deliveryIdString)
if err != nil {
return err
}
pickupPoint := messages.PickupPoint{}
err = json.NewDecoder(r.Body).Decode(&pickupPoint)
if err != nil {
return err
}
reply, err := s.pool.Process(id, Message{
Type: SetPickupPointType,
Content: &messages.SetPickupPoint{
DeliveryId: int64(deliveryId),
Id: pickupPoint.Id,
Name: pickupPoint.Name,
Address: pickupPoint.Address,
City: pickupPoint.City,
Zip: pickupPoint.Zip,
Country: pickupPoint.Country,
},
})
if err != nil {
return err
}
return s.WriteResult(w, reply)
}
func (s *PoolServer) HandleRemoveDelivery(w http.ResponseWriter, r *http.Request, id CartId) error {
deliveryIdString := r.PathValue("deliveryId")
deliveryId, err := strconv.Atoi(deliveryIdString)
if err != nil {
return err
}
reply, err := s.pool.Process(id, Message{
Type: RemoveDeliveryType,
Content: &messages.RemoveDelivery{Id: int64(deliveryId)},
})
if err != nil {
return err
}
return s.WriteResult(w, reply)
}
func (s *PoolServer) HandleQuantityChange(w http.ResponseWriter, r *http.Request, id CartId) error {
changeQuantity := messages.ChangeQuantity{}
err := json.NewDecoder(r.Body).Decode(&changeQuantity)
if err != nil {
return err
}
reply, err := s.pool.Process(id, Message{
Type: ChangeQuantityType,
Content: &changeQuantity,
})
if err != nil {
return err
}
return s.WriteResult(w, reply)
}
func (s *PoolServer) HandleSetCartItems(w http.ResponseWriter, r *http.Request, id CartId) error {
setCartItems := messages.SetCartRequest{}
err := json.NewDecoder(r.Body).Decode(&setCartItems)
if err != nil {
return err
}
reply, err := s.pool.Process(id, Message{
Type: SetCartItemsType,
Content: &setCartItems,
})
if err != nil {
return err
}
return s.WriteResult(w, reply)
}
func (s *PoolServer) HandleAddRequest(w http.ResponseWriter, r *http.Request, id CartId) error {
addRequest := messages.AddRequest{}
err := json.NewDecoder(r.Body).Decode(&addRequest)
if err != nil {
return err
}
reply, err := s.pool.Process(id, Message{
Type: AddRequestType,
Content: &addRequest,
})
if err != nil {
return err
}
return s.WriteResult(w, reply)
}
func (s *PoolServer) HandleConfirmation(w http.ResponseWriter, r *http.Request, id CartId) error {
orderId := r.PathValue("orderId")
if orderId == "" {
return fmt.Errorf("orderId is empty")
}
order, err := KlarnaInstance.GetOrder(orderId)
if err != nil {
return err
}
w.Header().Set("Content-Type", "application/json")
w.Header().Set("X-Pod-Name", s.pod_name)
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Access-Control-Allow-Origin", "*")
w.WriteHeader(http.StatusOK)
return json.NewEncoder(w).Encode(order)
}
func (s *PoolServer) HandleCheckout(w http.ResponseWriter, r *http.Request, id CartId) error {
reply, err := s.pool.Process(id, Message{
Type: CreateCheckoutOrderType,
Content: &messages.CreateCheckoutOrder{
Terms: "https://slask-finder.tornberg.me/terms",
Checkout: "https://slask-finder.tornberg.me/checkout?order_id={checkout.order.id}",
Confirmation: "https://slask-finder.tornberg.me/confirmation/{checkout.order.id}",
Push: "https://cart.tornberg.me/push?order_id={checkout.order.id}",
},
})
if err != nil {
return err
}
if reply.StatusCode != 200 {
return s.WriteResult(w, reply)
}
// w.Header().Set("Content-Type", "application/json")
// w.Header().Set("X-Pod-Name", s.pod_name)
// w.Header().Set("Cache-Control", "no-cache")
// w.Header().Set("Access-Control-Allow-Origin", "*")
// w.WriteHeader(http.StatusOK)
return s.WriteResult(w, reply)
}
func NewCartId() CartId {
id := time.Now().UnixNano() + rand.Int63()
return ToCartId(fmt.Sprintf("%d", id))
}
func CookieCartIdHandler(fn func(w http.ResponseWriter, r *http.Request, cartId CartId) error) func(w http.ResponseWriter, r *http.Request) error {
return func(w http.ResponseWriter, r *http.Request) error {
var cartId CartId
cartIdCookie := r.CookiesNamed("cartid")
if cartIdCookie == nil || len(cartIdCookie) == 0 {
cartId = NewCartId()
http.SetCookie(w, &http.Cookie{
Name: "cartid",
Value: cartId.String(),
Secure: true,
HttpOnly: true,
Path: "/",
Expires: time.Now().AddDate(0, 0, 14),
SameSite: http.SameSiteLaxMode,
})
} else {
cartId = ToCartId(cartIdCookie[0].Value)
}
return fn(w, r, cartId)
}
}
func (s *PoolServer) RemoveCartCookie(w http.ResponseWriter, r *http.Request, cartId CartId) error {
cartId = NewCartId()
http.SetCookie(w, &http.Cookie{
Name: "cartid",
Value: cartId.String(),
Path: "/",
Secure: true,
HttpOnly: true,
Expires: time.Unix(0, 0),
SameSite: http.SameSiteLaxMode,
})
w.WriteHeader(http.StatusOK)
return nil
}
func CartIdHandler(fn func(w http.ResponseWriter, r *http.Request, cartId CartId) error) func(w http.ResponseWriter, r *http.Request) error {
return func(w http.ResponseWriter, r *http.Request) error {
cartId := ToCartId(r.PathValue("id"))
return fn(w, r, cartId)
}
}
func (s *PoolServer) Serve() *http.ServeMux {
mux := http.NewServeMux()
//mux.HandleFunc("/", s.RewritePath)
mux.HandleFunc("OPTIONS /", func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Access-Control-Allow-Origin", "*")
w.Header().Set("Access-Control-Allow-Methods", "GET, PUT, POST, DELETE")
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
w.WriteHeader(http.StatusOK)
})
mux.HandleFunc("GET /", ErrorHandler(CookieCartIdHandler(s.HandleGet)))
mux.HandleFunc("GET /add/{sku}", ErrorHandler(CookieCartIdHandler(s.HandleAddSku)))
mux.HandleFunc("POST /", ErrorHandler(CookieCartIdHandler(s.HandleAddRequest)))
mux.HandleFunc("POST /set", ErrorHandler(CookieCartIdHandler(s.HandleSetCartItems)))
mux.HandleFunc("DELETE /{itemId}", ErrorHandler(CookieCartIdHandler(s.HandleDeleteItem)))
mux.HandleFunc("PUT /", ErrorHandler(CookieCartIdHandler(s.HandleQuantityChange)))
mux.HandleFunc("DELETE /", ErrorHandler(CookieCartIdHandler(s.RemoveCartCookie)))
mux.HandleFunc("POST /delivery", ErrorHandler(CookieCartIdHandler(s.HandleSetDelivery)))
mux.HandleFunc("DELETE /delivery/{deliveryId}", ErrorHandler(CookieCartIdHandler(s.HandleRemoveDelivery)))
mux.HandleFunc("PUT /delivery/{deliveryId}/pickupPoint", ErrorHandler(CookieCartIdHandler(s.HandleSetPickupPoint)))
mux.HandleFunc("GET /checkout", ErrorHandler(CookieCartIdHandler(s.HandleCheckout)))
mux.HandleFunc("GET /confirmation/{orderId}", ErrorHandler(CookieCartIdHandler(s.HandleConfirmation)))
mux.HandleFunc("GET /byid/{id}", ErrorHandler(CartIdHandler(s.HandleGet)))
mux.HandleFunc("GET /byid/{id}/add/{sku}", ErrorHandler(CartIdHandler(s.HandleAddSku)))
mux.HandleFunc("POST /byid/{id}", ErrorHandler(CartIdHandler(s.HandleAddRequest)))
mux.HandleFunc("DELETE /byid/{id}/{itemId}", ErrorHandler(CartIdHandler(s.HandleDeleteItem)))
mux.HandleFunc("PUT /byid/{id}", ErrorHandler(CartIdHandler(s.HandleQuantityChange)))
mux.HandleFunc("POST /byid/{id}/delivery", ErrorHandler(CartIdHandler(s.HandleSetDelivery)))
mux.HandleFunc("DELETE /byid/{id}/delivery/{deliveryId}", ErrorHandler(CartIdHandler(s.HandleRemoveDelivery)))
mux.HandleFunc("PUT /byid/{id}/delivery/{deliveryId}/pickupPoint", ErrorHandler(CartIdHandler(s.HandleSetPickupPoint)))
mux.HandleFunc("GET /byid/{id}/checkout", ErrorHandler(CartIdHandler(s.HandleCheckout)))
mux.HandleFunc("GET /byid/{id}/confirmation", ErrorHandler(CartIdHandler(s.HandleConfirmation)))
return mux
}
-35
View File
@@ -1,35 +0,0 @@
package main
import (
"encoding/json"
"fmt"
"net/http"
"github.com/matst80/slask-finder/pkg/index"
)
// TODO make this configurable
func getBaseUrl(country string) string {
// if country == "se" {
// return "http://s10n-se:8080"
// }
if country == "no" {
return "http://s10n-no.s10n:8080"
}
if country == "se" {
return "http://s10n-se.s10n:8080"
}
return "http://localhost:8082"
}
func FetchItem(sku string, country string) (*index.DataItem, error) {
baseUrl := getBaseUrl(country)
res, err := http.Get(fmt.Sprintf("%s/api/by-sku/%s", baseUrl, sku))
if err != nil {
return nil, err
}
defer res.Body.Close()
var item index.DataItem
err = json.NewDecoder(res.Body).Decode(&item)
return &item, err
}
-420
View File
@@ -1,420 +0,0 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.10
// protoc v3.21.12
// source: cart_actor.proto
package messages
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
unsafe "unsafe"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
// MutationType corresponds 1:1 with the legacy uint16 message type constants.
type MutationType int32
const (
MutationType_MUTATION_TYPE_UNSPECIFIED MutationType = 0
MutationType_MUTATION_ADD_REQUEST MutationType = 1
MutationType_MUTATION_ADD_ITEM MutationType = 2
// (3 was unused / reserved in legacy framing)
MutationType_MUTATION_REMOVE_ITEM MutationType = 4
MutationType_MUTATION_REMOVE_DELIVERY MutationType = 5
MutationType_MUTATION_CHANGE_QUANTITY MutationType = 6
MutationType_MUTATION_SET_DELIVERY MutationType = 7
MutationType_MUTATION_SET_PICKUP_POINT MutationType = 8
MutationType_MUTATION_CREATE_CHECKOUT_ORDER MutationType = 9
MutationType_MUTATION_SET_CART_ITEMS MutationType = 10
MutationType_MUTATION_ORDER_COMPLETED MutationType = 11
)
// Enum value maps for MutationType.
var (
MutationType_name = map[int32]string{
0: "MUTATION_TYPE_UNSPECIFIED",
1: "MUTATION_ADD_REQUEST",
2: "MUTATION_ADD_ITEM",
4: "MUTATION_REMOVE_ITEM",
5: "MUTATION_REMOVE_DELIVERY",
6: "MUTATION_CHANGE_QUANTITY",
7: "MUTATION_SET_DELIVERY",
8: "MUTATION_SET_PICKUP_POINT",
9: "MUTATION_CREATE_CHECKOUT_ORDER",
10: "MUTATION_SET_CART_ITEMS",
11: "MUTATION_ORDER_COMPLETED",
}
MutationType_value = map[string]int32{
"MUTATION_TYPE_UNSPECIFIED": 0,
"MUTATION_ADD_REQUEST": 1,
"MUTATION_ADD_ITEM": 2,
"MUTATION_REMOVE_ITEM": 4,
"MUTATION_REMOVE_DELIVERY": 5,
"MUTATION_CHANGE_QUANTITY": 6,
"MUTATION_SET_DELIVERY": 7,
"MUTATION_SET_PICKUP_POINT": 8,
"MUTATION_CREATE_CHECKOUT_ORDER": 9,
"MUTATION_SET_CART_ITEMS": 10,
"MUTATION_ORDER_COMPLETED": 11,
}
)
func (x MutationType) Enum() *MutationType {
p := new(MutationType)
*p = x
return p
}
func (x MutationType) String() string {
return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
}
func (MutationType) Descriptor() protoreflect.EnumDescriptor {
return file_cart_actor_proto_enumTypes[0].Descriptor()
}
func (MutationType) Type() protoreflect.EnumType {
return &file_cart_actor_proto_enumTypes[0]
}
func (x MutationType) Number() protoreflect.EnumNumber {
return protoreflect.EnumNumber(x)
}
// Deprecated: Use MutationType.Descriptor instead.
func (MutationType) EnumDescriptor() ([]byte, []int) {
return file_cart_actor_proto_rawDescGZIP(), []int{0}
}
// MutationRequest is an envelope:
// - cart_id: string form of CartId (legacy 16-byte array truncated/padded).
// - type: mutation kind (see enum).
// - payload: serialized underlying proto message (AddRequest, AddItem, etc.).
// - client_timestamp: optional unix timestamp; server sets if zero.
type MutationRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
CartId string `protobuf:"bytes,1,opt,name=cart_id,json=cartId,proto3" json:"cart_id,omitempty"`
Type MutationType `protobuf:"varint,2,opt,name=type,proto3,enum=messages.MutationType" json:"type,omitempty"`
Payload []byte `protobuf:"bytes,3,opt,name=payload,proto3" json:"payload,omitempty"`
ClientTimestamp int64 `protobuf:"varint,4,opt,name=client_timestamp,json=clientTimestamp,proto3" json:"client_timestamp,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *MutationRequest) Reset() {
*x = MutationRequest{}
mi := &file_cart_actor_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *MutationRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*MutationRequest) ProtoMessage() {}
func (x *MutationRequest) ProtoReflect() protoreflect.Message {
mi := &file_cart_actor_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use MutationRequest.ProtoReflect.Descriptor instead.
func (*MutationRequest) Descriptor() ([]byte, []int) {
return file_cart_actor_proto_rawDescGZIP(), []int{0}
}
func (x *MutationRequest) GetCartId() string {
if x != nil {
return x.CartId
}
return ""
}
func (x *MutationRequest) GetType() MutationType {
if x != nil {
return x.Type
}
return MutationType_MUTATION_TYPE_UNSPECIFIED
}
func (x *MutationRequest) GetPayload() []byte {
if x != nil {
return x.Payload
}
return nil
}
func (x *MutationRequest) GetClientTimestamp() int64 {
if x != nil {
return x.ClientTimestamp
}
return 0
}
// MutationReply returns a status code (legacy semantics) plus a JSON payload
// representing the full cart state (or an error message if non-200).
type MutationReply struct {
state protoimpl.MessageState `protogen:"open.v1"`
StatusCode int32 `protobuf:"varint,1,opt,name=status_code,json=statusCode,proto3" json:"status_code,omitempty"`
Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"` // JSON cart state or error string
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *MutationReply) Reset() {
*x = MutationReply{}
mi := &file_cart_actor_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *MutationReply) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*MutationReply) ProtoMessage() {}
func (x *MutationReply) ProtoReflect() protoreflect.Message {
mi := &file_cart_actor_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use MutationReply.ProtoReflect.Descriptor instead.
func (*MutationReply) Descriptor() ([]byte, []int) {
return file_cart_actor_proto_rawDescGZIP(), []int{1}
}
func (x *MutationReply) GetStatusCode() int32 {
if x != nil {
return x.StatusCode
}
return 0
}
func (x *MutationReply) GetPayload() []byte {
if x != nil {
return x.Payload
}
return nil
}
// StateRequest fetches current cart state without mutation.
type StateRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
CartId string `protobuf:"bytes,1,opt,name=cart_id,json=cartId,proto3" json:"cart_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *StateRequest) Reset() {
*x = StateRequest{}
mi := &file_cart_actor_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *StateRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*StateRequest) ProtoMessage() {}
func (x *StateRequest) ProtoReflect() protoreflect.Message {
mi := &file_cart_actor_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use StateRequest.ProtoReflect.Descriptor instead.
func (*StateRequest) Descriptor() ([]byte, []int) {
return file_cart_actor_proto_rawDescGZIP(), []int{2}
}
func (x *StateRequest) GetCartId() string {
if x != nil {
return x.CartId
}
return ""
}
// StateReply mirrors MutationReply for consistency.
type StateReply struct {
state protoimpl.MessageState `protogen:"open.v1"`
StatusCode int32 `protobuf:"varint,1,opt,name=status_code,json=statusCode,proto3" json:"status_code,omitempty"`
Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"` // JSON cart state or error string
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *StateReply) Reset() {
*x = StateReply{}
mi := &file_cart_actor_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *StateReply) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*StateReply) ProtoMessage() {}
func (x *StateReply) ProtoReflect() protoreflect.Message {
mi := &file_cart_actor_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use StateReply.ProtoReflect.Descriptor instead.
func (*StateReply) Descriptor() ([]byte, []int) {
return file_cart_actor_proto_rawDescGZIP(), []int{3}
}
func (x *StateReply) GetStatusCode() int32 {
if x != nil {
return x.StatusCode
}
return 0
}
func (x *StateReply) GetPayload() []byte {
if x != nil {
return x.Payload
}
return nil
}
var File_cart_actor_proto protoreflect.FileDescriptor
const file_cart_actor_proto_rawDesc = "" +
"\n" +
"\x10cart_actor.proto\x12\bmessages\"\x9b\x01\n" +
"\x0fMutationRequest\x12\x17\n" +
"\acart_id\x18\x01 \x01(\tR\x06cartId\x12*\n" +
"\x04type\x18\x02 \x01(\x0e2\x16.messages.MutationTypeR\x04type\x12\x18\n" +
"\apayload\x18\x03 \x01(\fR\apayload\x12)\n" +
"\x10client_timestamp\x18\x04 \x01(\x03R\x0fclientTimestamp\"J\n" +
"\rMutationReply\x12\x1f\n" +
"\vstatus_code\x18\x01 \x01(\x05R\n" +
"statusCode\x12\x18\n" +
"\apayload\x18\x02 \x01(\fR\apayload\"'\n" +
"\fStateRequest\x12\x17\n" +
"\acart_id\x18\x01 \x01(\tR\x06cartId\"G\n" +
"\n" +
"StateReply\x12\x1f\n" +
"\vstatus_code\x18\x01 \x01(\x05R\n" +
"statusCode\x12\x18\n" +
"\apayload\x18\x02 \x01(\fR\apayload*\xcd\x02\n" +
"\fMutationType\x12\x1d\n" +
"\x19MUTATION_TYPE_UNSPECIFIED\x10\x00\x12\x18\n" +
"\x14MUTATION_ADD_REQUEST\x10\x01\x12\x15\n" +
"\x11MUTATION_ADD_ITEM\x10\x02\x12\x18\n" +
"\x14MUTATION_REMOVE_ITEM\x10\x04\x12\x1c\n" +
"\x18MUTATION_REMOVE_DELIVERY\x10\x05\x12\x1c\n" +
"\x18MUTATION_CHANGE_QUANTITY\x10\x06\x12\x19\n" +
"\x15MUTATION_SET_DELIVERY\x10\a\x12\x1d\n" +
"\x19MUTATION_SET_PICKUP_POINT\x10\b\x12\"\n" +
"\x1eMUTATION_CREATE_CHECKOUT_ORDER\x10\t\x12\x1b\n" +
"\x17MUTATION_SET_CART_ITEMS\x10\n" +
"\x12\x1c\n" +
"\x18MUTATION_ORDER_COMPLETED\x10\v2\x83\x01\n" +
"\tCartActor\x12<\n" +
"\x06Mutate\x12\x19.messages.MutationRequest\x1a\x17.messages.MutationReply\x128\n" +
"\bGetState\x12\x16.messages.StateRequest\x1a\x14.messages.StateReplyB\fZ\n" +
".;messagesb\x06proto3"
var (
file_cart_actor_proto_rawDescOnce sync.Once
file_cart_actor_proto_rawDescData []byte
)
func file_cart_actor_proto_rawDescGZIP() []byte {
file_cart_actor_proto_rawDescOnce.Do(func() {
file_cart_actor_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_cart_actor_proto_rawDesc), len(file_cart_actor_proto_rawDesc)))
})
return file_cart_actor_proto_rawDescData
}
var file_cart_actor_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
var file_cart_actor_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
var file_cart_actor_proto_goTypes = []any{
(MutationType)(0), // 0: messages.MutationType
(*MutationRequest)(nil), // 1: messages.MutationRequest
(*MutationReply)(nil), // 2: messages.MutationReply
(*StateRequest)(nil), // 3: messages.StateRequest
(*StateReply)(nil), // 4: messages.StateReply
}
var file_cart_actor_proto_depIdxs = []int32{
0, // 0: messages.MutationRequest.type:type_name -> messages.MutationType
1, // 1: messages.CartActor.Mutate:input_type -> messages.MutationRequest
3, // 2: messages.CartActor.GetState:input_type -> messages.StateRequest
2, // 3: messages.CartActor.Mutate:output_type -> messages.MutationReply
4, // 4: messages.CartActor.GetState:output_type -> messages.StateReply
3, // [3:5] is the sub-list for method output_type
1, // [1:3] is the sub-list for method input_type
1, // [1:1] is the sub-list for extension type_name
1, // [1:1] is the sub-list for extension extendee
0, // [0:1] is the sub-list for field type_name
}
func init() { file_cart_actor_proto_init() }
func file_cart_actor_proto_init() {
if File_cart_actor_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_cart_actor_proto_rawDesc), len(file_cart_actor_proto_rawDesc)),
NumEnums: 1,
NumMessages: 4,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_cart_actor_proto_goTypes,
DependencyIndexes: file_cart_actor_proto_depIdxs,
EnumInfos: file_cart_actor_proto_enumTypes,
MessageInfos: file_cart_actor_proto_msgTypes,
}.Build()
File_cart_actor_proto = out.File
file_cart_actor_proto_goTypes = nil
file_cart_actor_proto_depIdxs = nil
}
-89
View File
@@ -1,89 +0,0 @@
syntax = "proto3";
package messages;
option go_package = ".;messages";
// -----------------------------------------------------------------------------
// Cart Actor gRPC API (Envelope Variant)
// -----------------------------------------------------------------------------
// This service replaces the legacy custom TCP frame protocol used on port 1337.
// It keeps the existing per-mutation proto messages (defined in messages.proto)
// serialized into an opaque `bytes payload` field for minimal refactor cost.
// The numeric values in MutationType MUST match the legacy message type
// constants (see message-types.go) so persisted event logs replay correctly.
// -----------------------------------------------------------------------------
// MutationType corresponds 1:1 with the legacy uint16 message type constants.
enum MutationType {
MUTATION_TYPE_UNSPECIFIED = 0;
MUTATION_ADD_REQUEST = 1;
MUTATION_ADD_ITEM = 2;
// (3 was unused / reserved in legacy framing)
MUTATION_REMOVE_ITEM = 4;
MUTATION_REMOVE_DELIVERY = 5;
MUTATION_CHANGE_QUANTITY = 6;
MUTATION_SET_DELIVERY = 7;
MUTATION_SET_PICKUP_POINT = 8;
MUTATION_CREATE_CHECKOUT_ORDER = 9;
MUTATION_SET_CART_ITEMS = 10;
MUTATION_ORDER_COMPLETED = 11;
}
// MutationRequest is an envelope:
// - cart_id: string form of CartId (legacy 16-byte array truncated/padded).
// - type: mutation kind (see enum).
// - payload: serialized underlying proto message (AddRequest, AddItem, etc.).
// - client_timestamp: optional unix timestamp; server sets if zero.
message MutationRequest {
string cart_id = 1;
MutationType type = 2;
bytes payload = 3;
int64 client_timestamp = 4;
}
// MutationReply returns a status code (legacy semantics) plus a JSON payload
// representing the full cart state (or an error message if non-200).
message MutationReply {
int32 status_code = 1;
bytes payload = 2; // JSON cart state or error string
}
// StateRequest fetches current cart state without mutation.
message StateRequest {
string cart_id = 1;
}
// StateReply mirrors MutationReply for consistency.
message StateReply {
int32 status_code = 1;
bytes payload = 2; // JSON cart state or error string
}
// CartActor exposes mutation and state retrieval for remote grains.
service CartActor {
// Mutate applies a single mutation to a cart, creating the cart lazily if needed.
rpc Mutate(MutationRequest) returns (MutationReply);
// GetState retrieves the cart's current state (JSON).
rpc GetState(StateRequest) returns (StateReply);
}
// -----------------------------------------------------------------------------
// Notes:
//
// 1. Generation:
// protoc --go_out=. --go_opt=paths=source_relative \
// --go-grpc_out=. --go-grpc_opt=paths=source_relative \
// cart_actor.proto
//
// 2. Underlying mutation payloads originate from messages.proto definitions.
// The server side will route based on MutationType and decode payload bytes
// using existing handler registry logic.
//
// 3. Future Enhancements:
// - Replace JSON state payload with a strongly typed CartState proto.
// - Add streaming RPC (e.g. WatchState) for live updates.
// - Migrate control plane (negotiate/ownership) into a separate proto
// (control_plane.proto) as per the migration plan.
// -----------------------------------------------------------------------------
-167
View File
@@ -1,167 +0,0 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.5.1
// - protoc v3.21.12
// source: cart_actor.proto
package messages
import (
context "context"
grpc "google.golang.org/grpc"
codes "google.golang.org/grpc/codes"
status "google.golang.org/grpc/status"
)
// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
// Requires gRPC-Go v1.64.0 or later.
const _ = grpc.SupportPackageIsVersion9
const (
CartActor_Mutate_FullMethodName = "/messages.CartActor/Mutate"
CartActor_GetState_FullMethodName = "/messages.CartActor/GetState"
)
// CartActorClient is the client API for CartActor service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
//
// CartActor exposes mutation and state retrieval for remote grains.
type CartActorClient interface {
// Mutate applies a single mutation to a cart, creating the cart lazily if needed.
Mutate(ctx context.Context, in *MutationRequest, opts ...grpc.CallOption) (*MutationReply, error)
// GetState retrieves the cart's current state (JSON).
GetState(ctx context.Context, in *StateRequest, opts ...grpc.CallOption) (*StateReply, error)
}
type cartActorClient struct {
cc grpc.ClientConnInterface
}
func NewCartActorClient(cc grpc.ClientConnInterface) CartActorClient {
return &cartActorClient{cc}
}
func (c *cartActorClient) Mutate(ctx context.Context, in *MutationRequest, opts ...grpc.CallOption) (*MutationReply, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(MutationReply)
err := c.cc.Invoke(ctx, CartActor_Mutate_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *cartActorClient) GetState(ctx context.Context, in *StateRequest, opts ...grpc.CallOption) (*StateReply, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(StateReply)
err := c.cc.Invoke(ctx, CartActor_GetState_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// CartActorServer is the server API for CartActor service.
// All implementations must embed UnimplementedCartActorServer
// for forward compatibility.
//
// CartActor exposes mutation and state retrieval for remote grains.
type CartActorServer interface {
// Mutate applies a single mutation to a cart, creating the cart lazily if needed.
Mutate(context.Context, *MutationRequest) (*MutationReply, error)
// GetState retrieves the cart's current state (JSON).
GetState(context.Context, *StateRequest) (*StateReply, error)
mustEmbedUnimplementedCartActorServer()
}
// UnimplementedCartActorServer must be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedCartActorServer struct{}
func (UnimplementedCartActorServer) Mutate(context.Context, *MutationRequest) (*MutationReply, error) {
return nil, status.Errorf(codes.Unimplemented, "method Mutate not implemented")
}
func (UnimplementedCartActorServer) GetState(context.Context, *StateRequest) (*StateReply, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetState not implemented")
}
func (UnimplementedCartActorServer) mustEmbedUnimplementedCartActorServer() {}
func (UnimplementedCartActorServer) testEmbeddedByValue() {}
// UnsafeCartActorServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to CartActorServer will
// result in compilation errors.
type UnsafeCartActorServer interface {
mustEmbedUnimplementedCartActorServer()
}
func RegisterCartActorServer(s grpc.ServiceRegistrar, srv CartActorServer) {
// If the following call pancis, it indicates UnimplementedCartActorServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&CartActor_ServiceDesc, srv)
}
func _CartActor_Mutate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(MutationRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(CartActorServer).Mutate(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: CartActor_Mutate_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(CartActorServer).Mutate(ctx, req.(*MutationRequest))
}
return interceptor(ctx, in, info, handler)
}
func _CartActor_GetState_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(StateRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(CartActorServer).GetState(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: CartActor_GetState_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(CartActorServer).GetState(ctx, req.(*StateRequest))
}
return interceptor(ctx, in, info, handler)
}
// CartActor_ServiceDesc is the grpc.ServiceDesc for CartActor service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var CartActor_ServiceDesc = grpc.ServiceDesc{
ServiceName: "messages.CartActor",
HandlerType: (*CartActorServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "Mutate",
Handler: _CartActor_Mutate_Handler,
},
{
MethodName: "GetState",
Handler: _CartActor_GetState_Handler,
},
},
Streams: []grpc.StreamDesc{},
Metadata: "cart_actor.proto",
}
-496
View File
@@ -1,496 +0,0 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.10
// protoc v3.21.12
// source: control_plane.proto
package messages
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
unsafe "unsafe"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
// Empty request placeholder (common pattern).
type Empty struct {
state protoimpl.MessageState `protogen:"open.v1"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *Empty) Reset() {
*x = Empty{}
mi := &file_control_plane_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *Empty) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*Empty) ProtoMessage() {}
func (x *Empty) ProtoReflect() protoreflect.Message {
mi := &file_control_plane_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use Empty.ProtoReflect.Descriptor instead.
func (*Empty) Descriptor() ([]byte, []int) {
return file_control_plane_proto_rawDescGZIP(), []int{0}
}
// Ping reply includes responding host and its current unix time (seconds).
type PingReply struct {
state protoimpl.MessageState `protogen:"open.v1"`
Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"`
UnixTime int64 `protobuf:"varint,2,opt,name=unix_time,json=unixTime,proto3" json:"unix_time,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *PingReply) Reset() {
*x = PingReply{}
mi := &file_control_plane_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *PingReply) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*PingReply) ProtoMessage() {}
func (x *PingReply) ProtoReflect() protoreflect.Message {
mi := &file_control_plane_proto_msgTypes[1]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use PingReply.ProtoReflect.Descriptor instead.
func (*PingReply) Descriptor() ([]byte, []int) {
return file_control_plane_proto_rawDescGZIP(), []int{1}
}
func (x *PingReply) GetHost() string {
if x != nil {
return x.Host
}
return ""
}
func (x *PingReply) GetUnixTime() int64 {
if x != nil {
return x.UnixTime
}
return 0
}
// NegotiateRequest carries the caller's full view of known hosts (including self).
type NegotiateRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
KnownHosts []string `protobuf:"bytes,1,rep,name=known_hosts,json=knownHosts,proto3" json:"known_hosts,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *NegotiateRequest) Reset() {
*x = NegotiateRequest{}
mi := &file_control_plane_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *NegotiateRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*NegotiateRequest) ProtoMessage() {}
func (x *NegotiateRequest) ProtoReflect() protoreflect.Message {
mi := &file_control_plane_proto_msgTypes[2]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use NegotiateRequest.ProtoReflect.Descriptor instead.
func (*NegotiateRequest) Descriptor() ([]byte, []int) {
return file_control_plane_proto_rawDescGZIP(), []int{2}
}
func (x *NegotiateRequest) GetKnownHosts() []string {
if x != nil {
return x.KnownHosts
}
return nil
}
// NegotiateReply returns the callee's healthy hosts (including itself).
type NegotiateReply struct {
state protoimpl.MessageState `protogen:"open.v1"`
Hosts []string `protobuf:"bytes,1,rep,name=hosts,proto3" json:"hosts,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *NegotiateReply) Reset() {
*x = NegotiateReply{}
mi := &file_control_plane_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *NegotiateReply) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*NegotiateReply) ProtoMessage() {}
func (x *NegotiateReply) ProtoReflect() protoreflect.Message {
mi := &file_control_plane_proto_msgTypes[3]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use NegotiateReply.ProtoReflect.Descriptor instead.
func (*NegotiateReply) Descriptor() ([]byte, []int) {
return file_control_plane_proto_rawDescGZIP(), []int{3}
}
func (x *NegotiateReply) GetHosts() []string {
if x != nil {
return x.Hosts
}
return nil
}
// CartIdsReply returns the list of cart IDs (string form) currently owned locally.
type CartIdsReply struct {
state protoimpl.MessageState `protogen:"open.v1"`
CartIds []string `protobuf:"bytes,1,rep,name=cart_ids,json=cartIds,proto3" json:"cart_ids,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CartIdsReply) Reset() {
*x = CartIdsReply{}
mi := &file_control_plane_proto_msgTypes[4]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *CartIdsReply) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*CartIdsReply) ProtoMessage() {}
func (x *CartIdsReply) ProtoReflect() protoreflect.Message {
mi := &file_control_plane_proto_msgTypes[4]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use CartIdsReply.ProtoReflect.Descriptor instead.
func (*CartIdsReply) Descriptor() ([]byte, []int) {
return file_control_plane_proto_rawDescGZIP(), []int{4}
}
func (x *CartIdsReply) GetCartIds() []string {
if x != nil {
return x.CartIds
}
return nil
}
// OwnerChangeRequest notifies peers that ownership of a cart moved (or is moving) to new_host.
type OwnerChangeRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
CartId string `protobuf:"bytes,1,opt,name=cart_id,json=cartId,proto3" json:"cart_id,omitempty"`
NewHost string `protobuf:"bytes,2,opt,name=new_host,json=newHost,proto3" json:"new_host,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *OwnerChangeRequest) Reset() {
*x = OwnerChangeRequest{}
mi := &file_control_plane_proto_msgTypes[5]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *OwnerChangeRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*OwnerChangeRequest) ProtoMessage() {}
func (x *OwnerChangeRequest) ProtoReflect() protoreflect.Message {
mi := &file_control_plane_proto_msgTypes[5]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use OwnerChangeRequest.ProtoReflect.Descriptor instead.
func (*OwnerChangeRequest) Descriptor() ([]byte, []int) {
return file_control_plane_proto_rawDescGZIP(), []int{5}
}
func (x *OwnerChangeRequest) GetCartId() string {
if x != nil {
return x.CartId
}
return ""
}
func (x *OwnerChangeRequest) GetNewHost() string {
if x != nil {
return x.NewHost
}
return ""
}
// OwnerChangeAck indicates acceptance or rejection of an ownership change.
type OwnerChangeAck struct {
state protoimpl.MessageState `protogen:"open.v1"`
Accepted bool `protobuf:"varint,1,opt,name=accepted,proto3" json:"accepted,omitempty"`
Message string `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *OwnerChangeAck) Reset() {
*x = OwnerChangeAck{}
mi := &file_control_plane_proto_msgTypes[6]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *OwnerChangeAck) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*OwnerChangeAck) ProtoMessage() {}
func (x *OwnerChangeAck) ProtoReflect() protoreflect.Message {
mi := &file_control_plane_proto_msgTypes[6]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use OwnerChangeAck.ProtoReflect.Descriptor instead.
func (*OwnerChangeAck) Descriptor() ([]byte, []int) {
return file_control_plane_proto_rawDescGZIP(), []int{6}
}
func (x *OwnerChangeAck) GetAccepted() bool {
if x != nil {
return x.Accepted
}
return false
}
func (x *OwnerChangeAck) GetMessage() string {
if x != nil {
return x.Message
}
return ""
}
// ClosingNotice notifies peers this host is terminating (so they can drop / re-resolve).
type ClosingNotice struct {
state protoimpl.MessageState `protogen:"open.v1"`
Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ClosingNotice) Reset() {
*x = ClosingNotice{}
mi := &file_control_plane_proto_msgTypes[7]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ClosingNotice) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ClosingNotice) ProtoMessage() {}
func (x *ClosingNotice) ProtoReflect() protoreflect.Message {
mi := &file_control_plane_proto_msgTypes[7]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use ClosingNotice.ProtoReflect.Descriptor instead.
func (*ClosingNotice) Descriptor() ([]byte, []int) {
return file_control_plane_proto_rawDescGZIP(), []int{7}
}
func (x *ClosingNotice) GetHost() string {
if x != nil {
return x.Host
}
return ""
}
var File_control_plane_proto protoreflect.FileDescriptor
const file_control_plane_proto_rawDesc = "" +
"\n" +
"\x13control_plane.proto\x12\bmessages\"\a\n" +
"\x05Empty\"<\n" +
"\tPingReply\x12\x12\n" +
"\x04host\x18\x01 \x01(\tR\x04host\x12\x1b\n" +
"\tunix_time\x18\x02 \x01(\x03R\bunixTime\"3\n" +
"\x10NegotiateRequest\x12\x1f\n" +
"\vknown_hosts\x18\x01 \x03(\tR\n" +
"knownHosts\"&\n" +
"\x0eNegotiateReply\x12\x14\n" +
"\x05hosts\x18\x01 \x03(\tR\x05hosts\")\n" +
"\fCartIdsReply\x12\x19\n" +
"\bcart_ids\x18\x01 \x03(\tR\acartIds\"H\n" +
"\x12OwnerChangeRequest\x12\x17\n" +
"\acart_id\x18\x01 \x01(\tR\x06cartId\x12\x19\n" +
"\bnew_host\x18\x02 \x01(\tR\anewHost\"F\n" +
"\x0eOwnerChangeAck\x12\x1a\n" +
"\baccepted\x18\x01 \x01(\bR\baccepted\x12\x18\n" +
"\amessage\x18\x02 \x01(\tR\amessage\"#\n" +
"\rClosingNotice\x12\x12\n" +
"\x04host\x18\x01 \x01(\tR\x04host2\xbc\x02\n" +
"\fControlPlane\x12,\n" +
"\x04Ping\x12\x0f.messages.Empty\x1a\x13.messages.PingReply\x12A\n" +
"\tNegotiate\x12\x1a.messages.NegotiateRequest\x1a\x18.messages.NegotiateReply\x125\n" +
"\n" +
"GetCartIds\x12\x0f.messages.Empty\x1a\x16.messages.CartIdsReply\x12F\n" +
"\fConfirmOwner\x12\x1c.messages.OwnerChangeRequest\x1a\x18.messages.OwnerChangeAck\x12<\n" +
"\aClosing\x12\x17.messages.ClosingNotice\x1a\x18.messages.OwnerChangeAckB\fZ\n" +
".;messagesb\x06proto3"
var (
file_control_plane_proto_rawDescOnce sync.Once
file_control_plane_proto_rawDescData []byte
)
func file_control_plane_proto_rawDescGZIP() []byte {
file_control_plane_proto_rawDescOnce.Do(func() {
file_control_plane_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_control_plane_proto_rawDesc), len(file_control_plane_proto_rawDesc)))
})
return file_control_plane_proto_rawDescData
}
var file_control_plane_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
var file_control_plane_proto_goTypes = []any{
(*Empty)(nil), // 0: messages.Empty
(*PingReply)(nil), // 1: messages.PingReply
(*NegotiateRequest)(nil), // 2: messages.NegotiateRequest
(*NegotiateReply)(nil), // 3: messages.NegotiateReply
(*CartIdsReply)(nil), // 4: messages.CartIdsReply
(*OwnerChangeRequest)(nil), // 5: messages.OwnerChangeRequest
(*OwnerChangeAck)(nil), // 6: messages.OwnerChangeAck
(*ClosingNotice)(nil), // 7: messages.ClosingNotice
}
var file_control_plane_proto_depIdxs = []int32{
0, // 0: messages.ControlPlane.Ping:input_type -> messages.Empty
2, // 1: messages.ControlPlane.Negotiate:input_type -> messages.NegotiateRequest
0, // 2: messages.ControlPlane.GetCartIds:input_type -> messages.Empty
5, // 3: messages.ControlPlane.ConfirmOwner:input_type -> messages.OwnerChangeRequest
7, // 4: messages.ControlPlane.Closing:input_type -> messages.ClosingNotice
1, // 5: messages.ControlPlane.Ping:output_type -> messages.PingReply
3, // 6: messages.ControlPlane.Negotiate:output_type -> messages.NegotiateReply
4, // 7: messages.ControlPlane.GetCartIds:output_type -> messages.CartIdsReply
6, // 8: messages.ControlPlane.ConfirmOwner:output_type -> messages.OwnerChangeAck
6, // 9: messages.ControlPlane.Closing:output_type -> messages.OwnerChangeAck
5, // [5:10] is the sub-list for method output_type
0, // [0:5] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name
0, // [0:0] is the sub-list for extension extendee
0, // [0:0] is the sub-list for field type_name
}
func init() { file_control_plane_proto_init() }
func file_control_plane_proto_init() {
if File_control_plane_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_control_plane_proto_rawDesc), len(file_control_plane_proto_rawDesc)),
NumEnums: 0,
NumMessages: 8,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_control_plane_proto_goTypes,
DependencyIndexes: file_control_plane_proto_depIdxs,
MessageInfos: file_control_plane_proto_msgTypes,
}.Build()
File_control_plane_proto = out.File
file_control_plane_proto_goTypes = nil
file_control_plane_proto_depIdxs = nil
}
-89
View File
@@ -1,89 +0,0 @@
syntax = "proto3";
package messages;
option go_package = ".;messages";
// -----------------------------------------------------------------------------
// Control Plane gRPC API
// -----------------------------------------------------------------------------
// Replaces the legacy custom frame-based control channel (previously port 1338).
// Responsibilities:
// - Liveness (Ping)
// - Membership negotiation (Negotiate)
// - Cart ownership change propagation (ConfirmOwner)
// - Cart ID listing for remote grain spawning (GetCartIds)
// - Graceful shutdown notifications (Closing)
// No authentication / TLS is defined initially (can be added later).
// -----------------------------------------------------------------------------
// Empty request placeholder (common pattern).
message Empty {}
// Ping reply includes responding host and its current unix time (seconds).
message PingReply {
string host = 1;
int64 unix_time = 2;
}
// NegotiateRequest carries the caller's full view of known hosts (including self).
message NegotiateRequest {
repeated string known_hosts = 1;
}
// NegotiateReply returns the callee's healthy hosts (including itself).
message NegotiateReply {
repeated string hosts = 1;
}
// CartIdsReply returns the list of cart IDs (string form) currently owned locally.
message CartIdsReply {
repeated string cart_ids = 1;
}
// OwnerChangeRequest notifies peers that ownership of a cart moved (or is moving) to new_host.
message OwnerChangeRequest {
string cart_id = 1;
string new_host = 2;
}
// OwnerChangeAck indicates acceptance or rejection of an ownership change.
message OwnerChangeAck {
bool accepted = 1;
string message = 2;
}
// ClosingNotice notifies peers this host is terminating (so they can drop / re-resolve).
message ClosingNotice {
string host = 1;
}
// ControlPlane defines cluster coordination and ownership operations.
service ControlPlane {
// Ping for liveness; lightweight health signal.
rpc Ping(Empty) returns (PingReply);
// Negotiate merges host views; used during discovery & convergence.
rpc Negotiate(NegotiateRequest) returns (NegotiateReply);
// GetCartIds lists currently owned cart IDs on this node.
rpc GetCartIds(Empty) returns (CartIdsReply);
// ConfirmOwner announces/asks peers to acknowledge ownership transfer.
rpc ConfirmOwner(OwnerChangeRequest) returns (OwnerChangeAck);
// Closing announces graceful shutdown so peers can proactively adjust.
rpc Closing(ClosingNotice) returns (OwnerChangeAck);
}
// -----------------------------------------------------------------------------
// Generation Instructions:
// protoc --go_out=. --go_opt=paths=source_relative \
// --go-grpc_out=. --go-grpc_opt=paths=source_relative \
// control_plane.proto
//
// Future Enhancements:
// - Add a streaming membership watch (server -> client) for immediate updates.
// - Add TLS / mTLS for secure intra-cluster communication.
// - Add richer health metadata (load, grain count) in PingReply.
// -----------------------------------------------------------------------------
-287
View File
@@ -1,287 +0,0 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.5.1
// - protoc v3.21.12
// source: control_plane.proto
package messages
import (
context "context"
grpc "google.golang.org/grpc"
codes "google.golang.org/grpc/codes"
status "google.golang.org/grpc/status"
)
// This is a compile-time assertion to ensure that this generated file
// is compatible with the grpc package it is being compiled against.
// Requires gRPC-Go v1.64.0 or later.
const _ = grpc.SupportPackageIsVersion9
const (
ControlPlane_Ping_FullMethodName = "/messages.ControlPlane/Ping"
ControlPlane_Negotiate_FullMethodName = "/messages.ControlPlane/Negotiate"
ControlPlane_GetCartIds_FullMethodName = "/messages.ControlPlane/GetCartIds"
ControlPlane_ConfirmOwner_FullMethodName = "/messages.ControlPlane/ConfirmOwner"
ControlPlane_Closing_FullMethodName = "/messages.ControlPlane/Closing"
)
// ControlPlaneClient is the client API for ControlPlane service.
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
//
// ControlPlane defines cluster coordination and ownership operations.
type ControlPlaneClient interface {
// Ping for liveness; lightweight health signal.
Ping(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*PingReply, error)
// Negotiate merges host views; used during discovery & convergence.
Negotiate(ctx context.Context, in *NegotiateRequest, opts ...grpc.CallOption) (*NegotiateReply, error)
// GetCartIds lists currently owned cart IDs on this node.
GetCartIds(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*CartIdsReply, error)
// ConfirmOwner announces/asks peers to acknowledge ownership transfer.
ConfirmOwner(ctx context.Context, in *OwnerChangeRequest, opts ...grpc.CallOption) (*OwnerChangeAck, error)
// Closing announces graceful shutdown so peers can proactively adjust.
Closing(ctx context.Context, in *ClosingNotice, opts ...grpc.CallOption) (*OwnerChangeAck, error)
}
type controlPlaneClient struct {
cc grpc.ClientConnInterface
}
func NewControlPlaneClient(cc grpc.ClientConnInterface) ControlPlaneClient {
return &controlPlaneClient{cc}
}
func (c *controlPlaneClient) Ping(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*PingReply, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(PingReply)
err := c.cc.Invoke(ctx, ControlPlane_Ping_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *controlPlaneClient) Negotiate(ctx context.Context, in *NegotiateRequest, opts ...grpc.CallOption) (*NegotiateReply, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(NegotiateReply)
err := c.cc.Invoke(ctx, ControlPlane_Negotiate_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *controlPlaneClient) GetCartIds(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*CartIdsReply, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(CartIdsReply)
err := c.cc.Invoke(ctx, ControlPlane_GetCartIds_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *controlPlaneClient) ConfirmOwner(ctx context.Context, in *OwnerChangeRequest, opts ...grpc.CallOption) (*OwnerChangeAck, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(OwnerChangeAck)
err := c.cc.Invoke(ctx, ControlPlane_ConfirmOwner_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *controlPlaneClient) Closing(ctx context.Context, in *ClosingNotice, opts ...grpc.CallOption) (*OwnerChangeAck, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(OwnerChangeAck)
err := c.cc.Invoke(ctx, ControlPlane_Closing_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// ControlPlaneServer is the server API for ControlPlane service.
// All implementations must embed UnimplementedControlPlaneServer
// for forward compatibility.
//
// ControlPlane defines cluster coordination and ownership operations.
type ControlPlaneServer interface {
// Ping for liveness; lightweight health signal.
Ping(context.Context, *Empty) (*PingReply, error)
// Negotiate merges host views; used during discovery & convergence.
Negotiate(context.Context, *NegotiateRequest) (*NegotiateReply, error)
// GetCartIds lists currently owned cart IDs on this node.
GetCartIds(context.Context, *Empty) (*CartIdsReply, error)
// ConfirmOwner announces/asks peers to acknowledge ownership transfer.
ConfirmOwner(context.Context, *OwnerChangeRequest) (*OwnerChangeAck, error)
// Closing announces graceful shutdown so peers can proactively adjust.
Closing(context.Context, *ClosingNotice) (*OwnerChangeAck, error)
mustEmbedUnimplementedControlPlaneServer()
}
// UnimplementedControlPlaneServer must be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedControlPlaneServer struct{}
func (UnimplementedControlPlaneServer) Ping(context.Context, *Empty) (*PingReply, error) {
return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
}
func (UnimplementedControlPlaneServer) Negotiate(context.Context, *NegotiateRequest) (*NegotiateReply, error) {
return nil, status.Errorf(codes.Unimplemented, "method Negotiate not implemented")
}
func (UnimplementedControlPlaneServer) GetCartIds(context.Context, *Empty) (*CartIdsReply, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetCartIds not implemented")
}
func (UnimplementedControlPlaneServer) ConfirmOwner(context.Context, *OwnerChangeRequest) (*OwnerChangeAck, error) {
return nil, status.Errorf(codes.Unimplemented, "method ConfirmOwner not implemented")
}
func (UnimplementedControlPlaneServer) Closing(context.Context, *ClosingNotice) (*OwnerChangeAck, error) {
return nil, status.Errorf(codes.Unimplemented, "method Closing not implemented")
}
func (UnimplementedControlPlaneServer) mustEmbedUnimplementedControlPlaneServer() {}
func (UnimplementedControlPlaneServer) testEmbeddedByValue() {}
// UnsafeControlPlaneServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to ControlPlaneServer will
// result in compilation errors.
type UnsafeControlPlaneServer interface {
mustEmbedUnimplementedControlPlaneServer()
}
func RegisterControlPlaneServer(s grpc.ServiceRegistrar, srv ControlPlaneServer) {
// If the following call pancis, it indicates UnimplementedControlPlaneServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&ControlPlane_ServiceDesc, srv)
}
func _ControlPlane_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(Empty)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ControlPlaneServer).Ping(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: ControlPlane_Ping_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ControlPlaneServer).Ping(ctx, req.(*Empty))
}
return interceptor(ctx, in, info, handler)
}
func _ControlPlane_Negotiate_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(NegotiateRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ControlPlaneServer).Negotiate(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: ControlPlane_Negotiate_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ControlPlaneServer).Negotiate(ctx, req.(*NegotiateRequest))
}
return interceptor(ctx, in, info, handler)
}
func _ControlPlane_GetCartIds_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(Empty)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ControlPlaneServer).GetCartIds(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: ControlPlane_GetCartIds_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ControlPlaneServer).GetCartIds(ctx, req.(*Empty))
}
return interceptor(ctx, in, info, handler)
}
func _ControlPlane_ConfirmOwner_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(OwnerChangeRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ControlPlaneServer).ConfirmOwner(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: ControlPlane_ConfirmOwner_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ControlPlaneServer).ConfirmOwner(ctx, req.(*OwnerChangeRequest))
}
return interceptor(ctx, in, info, handler)
}
func _ControlPlane_Closing_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ClosingNotice)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(ControlPlaneServer).Closing(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: ControlPlane_Closing_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(ControlPlaneServer).Closing(ctx, req.(*ClosingNotice))
}
return interceptor(ctx, in, info, handler)
}
// ControlPlane_ServiceDesc is the grpc.ServiceDesc for ControlPlane service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var ControlPlane_ServiceDesc = grpc.ServiceDesc{
ServiceName: "messages.ControlPlane",
HandlerType: (*ControlPlaneServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "Ping",
Handler: _ControlPlane_Ping_Handler,
},
{
MethodName: "Negotiate",
Handler: _ControlPlane_Negotiate_Handler,
},
{
MethodName: "GetCartIds",
Handler: _ControlPlane_GetCartIds_Handler,
},
{
MethodName: "ConfirmOwner",
Handler: _ControlPlane_ConfirmOwner_Handler,
},
{
MethodName: "Closing",
Handler: _ControlPlane_Closing_Handler,
},
},
Streams: []grpc.StreamDesc{},
Metadata: "control_plane.proto",
}
+65 -920
View File
File diff suppressed because it is too large Load Diff
+6 -83
View File
@@ -3,93 +3,16 @@ package messages;
option go_package = ".;messages"; option go_package = ".;messages";
message AddRequest { message AddRequest {
int32 quantity = 1; string Sku = 2;
string sku = 2;
string country = 3;
optional string storeId = 4;
}
message SetCartRequest {
repeated AddRequest items = 1;
} }
message AddItem { message AddItem {
int64 item_id = 1; int32 Quantity = 2;
int32 quantity = 2; int64 Price = 3;
int64 price = 3; string Sku = 4;
int64 orgPrice = 9; string Name = 5;
string sku = 4; string Image = 6;
string name = 5;
string image = 6;
int32 stock = 7;
int32 tax = 8;
string brand = 13;
string category = 14;
string category2 = 15;
string category3 = 16;
string category4 = 17;
string category5 = 18;
string disclaimer = 10;
string articleType = 11;
string sellerId = 19;
string sellerName = 20;
string country = 21;
optional string outlet = 12;
optional string storeId = 22;
} }
message RemoveItem {
int64 Id = 1;
}
message ChangeQuantity {
int64 id = 1;
int32 quantity = 2;
}
message SetDelivery {
string provider = 1;
repeated int64 items = 2;
optional PickupPoint pickupPoint = 3;
string country = 4;
string zip = 5;
optional string address = 6;
optional string city = 7;
}
message SetPickupPoint {
int64 deliveryId = 1;
string id = 2;
optional string name = 3;
optional string address = 4;
optional string city = 5;
optional string zip = 6;
optional string country = 7;
}
message PickupPoint {
string id = 1;
optional string name = 2;
optional string address = 3;
optional string city = 4;
optional string zip = 5;
optional string country = 6;
}
message RemoveDelivery {
int64 id = 1;
}
message CreateCheckoutOrder {
string terms = 1;
string checkout = 2;
string confirmation = 3;
string push = 4;
string validation = 5;
string country = 6;
}
message OrderCreated {
string orderId = 1;
string status = 2;
}
View File
-67
View File
@@ -1,67 +0,0 @@
// package main
// import "sync"
// type RemoteGrainPool struct {
// mu sync.RWMutex
// Host string
// grains map[CartId]*RemoteGrain
// }
// func NewRemoteGrainPool(addr string) *RemoteGrainPool {
// return &RemoteGrainPool{
// Host: addr,
// grains: make(map[CartId]*RemoteGrain),
// }
// }
// func (p *RemoteGrainPool) findRemoteGrain(id CartId) *RemoteGrain {
// p.mu.RLock()
// grain, ok := p.grains[id]
// p.mu.RUnlock()
// if !ok {
// return nil
// }
// return grain
// }
// func (p *RemoteGrainPool) findOrCreateGrain(id CartId) (*RemoteGrain, error) {
// grain := p.findRemoteGrain(id)
// if grain == nil {
// grain, err := NewRemoteGrain(id, p.Host)
// if err != nil {
// return nil, err
// }
// p.mu.Lock()
// p.grains[id] = grain
// p.mu.Unlock()
// }
// return grain, nil
// }
// func (p *RemoteGrainPool) Delete(id CartId) {
// p.mu.Lock()
// delete(p.grains, id)
// p.mu.Unlock()
// }
// func (p *RemoteGrainPool) Process(id CartId, messages ...Message) (*FrameWithPayload, error) {
// var result *FrameWithPayload
// grain, err := p.findOrCreateGrain(id)
// if err != nil {
// return nil, err
// }
// for _, message := range messages {
// result, err = grain.HandleMessage(&message, false)
// }
// return result, err
// }
// func (p *RemoteGrainPool) Get(id CartId) (*FrameWithPayload, error) {
// grain, err := p.findOrCreateGrain(id)
// if err != nil {
// return nil, err
// }
// return grain.GetCurrentState()
// }
-147
View File
@@ -1,147 +0,0 @@
package main
import (
"bytes"
"context"
"fmt"
"log"
"time"
proto "git.tornberg.me/go-cart-actor/proto" // generated package name is 'messages'; aliased as proto for consistency
"google.golang.org/grpc"
)
// RemoteGrainGRPC is the gRPC-backed implementation of a remote grain.
// It mirrors the previous RemoteGrain (TCP/frame based) while using the
// new CartActor gRPC service. It implements the Grain interface so that
// SyncedPool can remain largely unchanged when swapping transport layers.
type RemoteGrainGRPC struct {
Id CartId
Host string
client proto.CartActorClient
// Optional: keep the underlying conn so higher-level code can close if needed
conn *grpc.ClientConn
// Per-call timeout settings (tunable)
mutateTimeout time.Duration
stateTimeout time.Duration
}
// NewRemoteGrainGRPC constructs a remote grain adapter from an existing gRPC client.
func NewRemoteGrainGRPC(id CartId, host string, client proto.CartActorClient) *RemoteGrainGRPC {
return &RemoteGrainGRPC{
Id: id,
Host: host,
client: client,
mutateTimeout: 800 * time.Millisecond,
stateTimeout: 400 * time.Millisecond,
}
}
// NewRemoteGrainGRPCWithConn dials the target and creates the gRPC client.
// target should be host:port (where the CartActor service is exposed).
func NewRemoteGrainGRPCWithConn(id CartId, host string, target string, dialOpts ...grpc.DialOption) (*RemoteGrainGRPC, error) {
// NOTE: insecure for initial migration; should be replaced with TLS later.
baseOpts := []grpc.DialOption{grpc.WithInsecure(), grpc.WithBlock()}
baseOpts = append(baseOpts, dialOpts...)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
conn, err := grpc.DialContext(ctx, target, baseOpts...)
if err != nil {
return nil, err
}
client := proto.NewCartActorClient(conn)
return &RemoteGrainGRPC{
Id: id,
Host: host,
client: client,
conn: conn,
mutateTimeout: 800 * time.Millisecond,
stateTimeout: 400 * time.Millisecond,
}, nil
}
func (g *RemoteGrainGRPC) GetId() CartId {
return g.Id
}
// HandleMessage serializes the underlying mutation proto (without legacy message header)
// and invokes the CartActor.Mutate RPC. It wraps the reply into a FrameWithPayload
// for compatibility with existing higher-level code paths.
func (g *RemoteGrainGRPC) HandleMessage(message *Message, isReplay bool) (*FrameWithPayload, error) {
if message == nil {
return nil, fmt.Errorf("nil message")
}
if isReplay {
// Remote replay not expected; ignore to keep parity with old implementation.
return nil, fmt.Errorf("replay not supported for remote grains")
}
handler, err := GetMessageHandler(message.Type)
if err != nil {
return nil, err
}
// Ensure timestamp set (legacy behavior)
if message.TimeStamp == nil {
ts := time.Now().Unix()
message.TimeStamp = &ts
}
// Marshal underlying proto payload only (no StorableMessageHeader)
var buf bytes.Buffer
err = handler.Write(message, &buf)
if err != nil {
return nil, fmt.Errorf("encode mutation payload: %w", err)
}
req := &proto.MutationRequest{
CartId: g.Id.String(),
Type: proto.MutationType(message.Type), // numeric mapping preserved
Payload: buf.Bytes(),
ClientTimestamp: *message.TimeStamp,
}
ctx, cancel := context.WithTimeout(context.Background(), g.mutateTimeout)
defer cancel()
resp, err := g.client.Mutate(ctx, req)
if err != nil {
return nil, err
}
frame := MakeFrameWithPayload(RemoteHandleMutationReply, StatusCode(resp.StatusCode), resp.Payload)
return &frame, nil
}
// GetCurrentState calls CartActor.GetState and returns a FrameWithPayload
// shaped like the legacy RemoteGetStateReply.
func (g *RemoteGrainGRPC) GetCurrentState() (*FrameWithPayload, error) {
ctx, cancel := context.WithTimeout(context.Background(), g.stateTimeout)
defer cancel()
resp, err := g.client.GetState(ctx, &proto.StateRequest{
CartId: g.Id.String(),
})
if err != nil {
return nil, err
}
frame := MakeFrameWithPayload(RemoteGetStateReply, StatusCode(resp.StatusCode), resp.Payload)
return &frame, nil
}
// Close closes the underlying gRPC connection if this adapter created it.
func (g *RemoteGrainGRPC) Close() error {
if g.conn != nil {
return g.conn.Close()
}
return nil
}
// Debug helper to log operations (optional).
func (g *RemoteGrainGRPC) logf(format string, args ...interface{}) {
log.Printf("[remote-grain-grpc host=%s id=%s] %s", g.Host, g.Id.String(), fmt.Sprintf(format, args...))
}
+110
View File
@@ -0,0 +1,110 @@
package main
import (
"io"
"net"
"strings"
)
type RemoteGrainPool struct {
Hosts []string
grains map[CartId]RemoteGrain
}
func (id CartId) String() string {
return strings.Trim(string(id[:]), "\x00")
}
func ToCartId(id string) CartId {
var result [16]byte
copy(result[:], []byte(id))
return result
}
type RemoteGrain struct {
client net.Conn
Id CartId
Address string
}
func NewRemoteGrain(id CartId, address string) *RemoteGrain {
return &RemoteGrain{
Id: id,
Address: address,
}
}
func (g *RemoteGrain) Connect() error {
if g.client == nil {
client, err := net.Dial("tcp", g.Address)
if err != nil {
return err
}
g.client = client
}
return nil
}
func (g *RemoteGrain) HandleMessage(message *Message, isReplay bool) ([]byte, error) {
err := SendCartPacket(g.client, g.Id, RemoteHandleMessage, message.Write)
if err != nil {
return nil, err
}
_, data, err := ReceivePacket(g.client)
return data, err
}
func (g *RemoteGrain) GetId() CartId {
return g.Id
}
func (g *RemoteGrain) GetCurrentState() ([]byte, error) {
err := SendCartPacket(g.client, g.Id, RemoteGetState, func(w io.Writer) error {
return nil
})
if err != nil {
return nil, err
}
_, data, err := ReceivePacket(g.client)
return data, err
}
func NewRemoteGrainPool(addr ...string) *RemoteGrainPool {
return &RemoteGrainPool{
Hosts: addr,
grains: make(map[CartId]RemoteGrain),
}
}
func (p *RemoteGrainPool) findRemoteGrain(id CartId) *RemoteGrain {
grain, ok := p.grains[id]
if !ok {
return nil
}
return &grain
}
func (p *RemoteGrainPool) Process(id CartId, messages ...Message) ([]byte, error) {
var result []byte
var err error
grain := p.findRemoteGrain(id)
if grain == nil {
grain = NewRemoteGrain(id, p.Hosts[0])
grain.Connect()
p.grains[id] = *grain
}
for _, message := range messages {
result, err = grain.HandleMessage(&message, false)
}
return result, err
}
func (p *RemoteGrainPool) Get(id CartId) ([]byte, error) {
grain := p.findRemoteGrain(id)
if grain == nil {
return nil, nil
}
return grain.GetCurrentState()
}
+114
View File
@@ -0,0 +1,114 @@
package main
import (
"encoding/binary"
"encoding/json"
"fmt"
"io"
"net"
)
type GrainHandler struct {
listener net.Listener
pool GrainPool
}
func (h *GrainHandler) GetState(id CartId, reply *Grain) error {
grain, err := h.pool.Get(id)
if err != nil {
return err
}
*reply = grain
return nil
}
func NewGrainHandler(pool GrainPool, listen string) (*GrainHandler, error) {
handler := &GrainHandler{
pool: pool,
}
l, err := net.Listen("tcp", listen)
handler.listener = l
return handler, err
}
func (h *GrainHandler) Serve() {
for {
// Accept incoming connections
conn, err := h.listener.Accept()
if err != nil {
fmt.Println("Error:", err)
continue
}
// Handle client connection in a goroutine
go h.handleClient(conn)
}
}
func (h *GrainHandler) handleClient(conn net.Conn) {
fmt.Println("Handling client connection")
defer conn.Close()
var packet CartPacket
for {
for {
err := binary.Read(conn, binary.LittleEndian, &packet)
if err != nil {
if err == io.EOF {
break
}
fmt.Println("Error reading packet:", err)
}
if packet.Version != 2 {
fmt.Printf("Unknown version %d", packet.Version)
break
}
switch packet.MessageType {
case RemoteHandleMessage:
fmt.Printf("Handling message\n")
var msg Message
err := MessageFromReader(conn, &msg)
if err != nil {
fmt.Println("Error reading message:", err)
}
fmt.Printf("Message: %s, %v\n", packet.Id.String(), msg)
grain, err := h.pool.Get(packet.Id)
if err != nil {
fmt.Println("Error getting grain:", err)
}
_, err = grain.HandleMessage(&msg, false)
if err != nil {
fmt.Println("Error handling message:", err)
}
SendPacket(conn, ResponseBody, func(w io.Writer) error {
data, err := json.Marshal(grain)
if err != nil {
return err
}
w.Write(data)
return nil
})
case RemoteGetState:
fmt.Printf("Package: %s %v\n", packet.Id.String(), packet)
grain, err := h.pool.Get(packet.Id)
if err != nil {
fmt.Println("Error getting grain:", err)
}
SendPacket(conn, ResponseBody, func(w io.Writer) error {
data, err := json.Marshal(grain)
if err != nil {
return err
}
w.Write(data)
return nil
})
}
}
}
}
+25
View File
@@ -0,0 +1,25 @@
package main
import "fmt"
type Registry interface {
Register(address string, id string) error
Get(id string) (*string, error)
}
type MemoryRegistry struct {
registry map[string]string
}
func (r *MemoryRegistry) Register(address string, id string) error {
r.registry[id] = address
return nil
}
func (r *MemoryRegistry) Get(id string) (*string, error) {
addr, ok := r.registry[id]
if !ok {
return nil, fmt.Errorf("id not found")
}
return &addr, nil
}
-473
View File
@@ -1,473 +0,0 @@
package main
import (
"context"
"fmt"
"log"
"sync"
"time"
proto "git.tornberg.me/go-cart-actor/proto"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"google.golang.org/grpc"
"k8s.io/apimachinery/pkg/watch"
)
// SyncedPool coordinates cart grain ownership across nodes using gRPC control plane
// and cart actor services. Legacy frame / TCP code has been removed.
//
// Responsibilities:
// - Local grain access (delegates to GrainLocalPool)
// - Remote grain proxy management (RemoteGrainGRPC)
// - Cluster membership (AddRemote via discovery + negotiation)
// - Ownership acquisition (quorum via ConfirmOwner RPC)
// - Health/ping monitoring & remote removal
//
// Thread-safety: public methods that mutate internal maps lock p.mu (RWMutex).
type SyncedPool struct {
Hostname string
local *GrainLocalPool
mu sync.RWMutex
// Remote host state (gRPC only)
remoteHosts map[string]*RemoteHostGRPC // host -> remote host
// Remote grain proxies (by cart id)
remoteIndex map[CartId]Grain
// Discovery handler for re-adding hosts after failures
discardedHostHandler *DiscardedHostHandler
// Metrics / instrumentation dependencies already declared globally
}
// RemoteHostGRPC tracks a remote host's clients & health.
type RemoteHostGRPC struct {
Host string
Conn *grpc.ClientConn
CartClient proto.CartActorClient
ControlClient proto.ControlPlaneClient
MissedPings int
}
func (r *RemoteHostGRPC) IsHealthy() bool {
return r.MissedPings < 3
}
var (
negotiationCount = promauto.NewCounter(prometheus.CounterOpts{
Name: "cart_remote_negotiation_total",
Help: "The total number of remote negotiations",
})
grainSyncCount = promauto.NewCounter(prometheus.CounterOpts{
Name: "cart_grain_sync_total",
Help: "The total number of grain owner changes",
})
connectedRemotes = promauto.NewGauge(prometheus.GaugeOpts{
Name: "cart_connected_remotes",
Help: "The number of connected remotes",
})
remoteLookupCount = promauto.NewCounter(prometheus.CounterOpts{
Name: "cart_remote_lookup_total",
Help: "The total number of remote lookups",
})
)
func NewSyncedPool(local *GrainLocalPool, hostname string, discovery Discovery) (*SyncedPool, error) {
p := &SyncedPool{
Hostname: hostname,
local: local,
remoteHosts: make(map[string]*RemoteHostGRPC),
remoteIndex: make(map[CartId]Grain),
discardedHostHandler: NewDiscardedHostHandler(1338),
}
p.discardedHostHandler.SetReconnectHandler(p.AddRemote)
if discovery != nil {
go func() {
time.Sleep(3 * time.Second) // allow gRPC server startup
log.Printf("Starting discovery watcher")
ch, err := discovery.Watch()
if err != nil {
log.Printf("Discovery error: %v", err)
return
}
for evt := range ch {
if evt.Host == "" {
continue
}
switch evt.Type {
case watch.Deleted:
if p.IsKnown(evt.Host) {
p.RemoveHost(evt.Host)
}
default:
if !p.IsKnown(evt.Host) {
log.Printf("Discovered host %s", evt.Host)
p.AddRemote(evt.Host)
}
}
}
}()
} else {
log.Printf("No discovery configured; expecting manual AddRemote or static host injection")
}
return p, nil
}
// ------------------------- Remote Host Management -----------------------------
// AddRemote dials a remote host and initializes grain proxies.
func (p *SyncedPool) AddRemote(host string) {
if host == "" || host == p.Hostname {
return
}
p.mu.Lock()
if _, exists := p.remoteHosts[host]; exists {
p.mu.Unlock()
return
}
p.mu.Unlock()
target := fmt.Sprintf("%s:1337", host)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
conn, err := grpc.DialContext(ctx, target, grpc.WithInsecure(), grpc.WithBlock())
if err != nil {
log.Printf("AddRemote: dial %s failed: %v", target, err)
return
}
cartClient := proto.NewCartActorClient(conn)
controlClient := proto.NewControlPlaneClient(conn)
// Health check (Ping) with limited retries
pings := 3
for pings > 0 {
ctxPing, cancelPing := context.WithTimeout(context.Background(), 1*time.Second)
_, pingErr := controlClient.Ping(ctxPing, &proto.Empty{})
cancelPing()
if pingErr == nil {
break
}
pings--
time.Sleep(200 * time.Millisecond)
if pings == 0 {
log.Printf("AddRemote: ping %s failed after retries: %v", host, pingErr)
conn.Close()
return
}
}
remote := &RemoteHostGRPC{
Host: host,
Conn: conn,
CartClient: cartClient,
ControlClient: controlClient,
MissedPings: 0,
}
p.mu.Lock()
p.remoteHosts[host] = remote
p.mu.Unlock()
connectedRemotes.Set(float64(p.RemoteCount()))
log.Printf("Connected to remote host %s", host)
go p.pingLoop(remote)
go p.initializeRemote(remote)
go p.Negotiate()
}
// initializeRemote fetches remote cart ids and sets up remote grain proxies.
func (p *SyncedPool) initializeRemote(remote *RemoteHostGRPC) {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
reply, err := remote.ControlClient.GetCartIds(ctx, &proto.Empty{})
if err != nil {
log.Printf("Init remote %s: GetCartIds error: %v", remote.Host, err)
return
}
count := 0
for _, idStr := range reply.CartIds {
if idStr == "" {
continue
}
p.SpawnRemoteGrain(ToCartId(idStr), remote.Host)
count++
}
log.Printf("Remote %s reported %d grains", remote.Host, count)
}
// RemoveHost removes remote host and its grains.
func (p *SyncedPool) RemoveHost(host string) {
p.mu.Lock()
remote, exists := p.remoteHosts[host]
if exists {
delete(p.remoteHosts, host)
}
// remove grains pointing to host
for id, g := range p.remoteIndex {
if rg, ok := g.(*RemoteGrainGRPC); ok && rg.Host == host {
delete(p.remoteIndex, id)
}
}
p.mu.Unlock()
if exists {
remote.Conn.Close()
}
connectedRemotes.Set(float64(p.RemoteCount()))
}
// RemoteCount returns number of tracked remote hosts.
func (p *SyncedPool) RemoteCount() int {
p.mu.RLock()
defer p.mu.RUnlock()
return len(p.remoteHosts)
}
func (p *SyncedPool) IsKnown(host string) bool {
if host == p.Hostname {
return true
}
p.mu.RLock()
defer p.mu.RUnlock()
_, ok := p.remoteHosts[host]
return ok
}
func (p *SyncedPool) ExcludeKnown(hosts []string) []string {
ret := make([]string, 0, len(hosts))
for _, h := range hosts {
if !p.IsKnown(h) {
ret = append(ret, h)
}
}
return ret
}
// ------------------------- Health / Ping -------------------------------------
func (p *SyncedPool) pingLoop(remote *RemoteHostGRPC) {
ticker := time.NewTicker(3 * time.Second)
defer ticker.Stop()
for range ticker.C {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
_, err := remote.ControlClient.Ping(ctx, &proto.Empty{})
cancel()
if err != nil {
remote.MissedPings++
log.Printf("Ping %s failed (%d)", remote.Host, remote.MissedPings)
if !remote.IsHealthy() {
log.Printf("Remote %s unhealthy, removing", remote.Host)
p.RemoveHost(remote.Host)
return
}
continue
}
remote.MissedPings = 0
}
}
func (p *SyncedPool) IsHealthy() bool {
p.mu.RLock()
defer p.mu.RUnlock()
for _, r := range p.remoteHosts {
if !r.IsHealthy() {
return false
}
}
return true
}
// ------------------------- Negotiation ---------------------------------------
func (p *SyncedPool) Negotiate() {
negotiationCount.Inc()
p.mu.RLock()
hosts := make([]string, 0, len(p.remoteHosts)+1)
hosts = append(hosts, p.Hostname)
for h := range p.remoteHosts {
hosts = append(hosts, h)
}
remotes := make([]*RemoteHostGRPC, 0, len(p.remoteHosts))
for _, r := range p.remoteHosts {
remotes = append(remotes, r)
}
p.mu.RUnlock()
for _, r := range remotes {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
reply, err := r.ControlClient.Negotiate(ctx, &proto.NegotiateRequest{KnownHosts: hosts})
cancel()
if err != nil {
log.Printf("Negotiate with %s failed: %v", r.Host, err)
continue
}
for _, h := range reply.Hosts {
if !p.IsKnown(h) {
p.AddRemote(h)
}
}
}
}
// ------------------------- Grain Management ----------------------------------
// RemoveRemoteGrain removes a remote grain mapping.
func (p *SyncedPool) RemoveRemoteGrain(id CartId) {
p.mu.Lock()
delete(p.remoteIndex, id)
p.mu.Unlock()
}
// SpawnRemoteGrain creates/updates a remote grain proxy for a given host.
func (p *SyncedPool) SpawnRemoteGrain(id CartId, host string) {
if id.String() == "" {
return
}
p.mu.Lock()
// If local grain exists, remove it (ownership changed)
if g, ok := p.local.grains[id]; ok && g != nil {
delete(p.local.grains, id)
}
remoteHost, ok := p.remoteHosts[host]
if !ok {
p.mu.Unlock()
log.Printf("SpawnRemoteGrain: host %s unknown (id=%s), attempting AddRemote", host, id)
go p.AddRemote(host)
return
}
rg := NewRemoteGrainGRPC(id, host, remoteHost.CartClient)
p.remoteIndex[id] = rg
p.mu.Unlock()
}
// GetHealthyRemotes returns a copy slice of healthy remote hosts.
func (p *SyncedPool) GetHealthyRemotes() []*RemoteHostGRPC {
p.mu.RLock()
defer p.mu.RUnlock()
ret := make([]*RemoteHostGRPC, 0, len(p.remoteHosts))
for _, r := range p.remoteHosts {
if r.IsHealthy() {
ret = append(ret, r)
}
}
return ret
}
// RequestOwnership attempts to become owner of a cart, requiring quorum.
// On success local grain is (or will be) created; peers spawn remote proxies.
func (p *SyncedPool) RequestOwnership(id CartId) error {
ok := 0
all := 0
remotes := p.GetHealthyRemotes()
for _, r := range remotes {
ctx, cancel := context.WithTimeout(context.Background(), 800*time.Millisecond)
reply, err := r.ControlClient.ConfirmOwner(ctx, &proto.OwnerChangeRequest{
CartId: id.String(),
NewHost: p.Hostname,
})
cancel()
all++
if err != nil || reply == nil || !reply.Accepted {
log.Printf("ConfirmOwner failure from %s for %s: %v (reply=%v)", r.Host, id, err, reply)
continue
}
ok++
}
// Quorum rule mirrors legacy:
// - If fewer than 3 total, require all.
// - Else require majority (ok >= all/2).
if (all < 3 && ok < all) || ok < (all/2) {
p.removeLocalGrain(id)
return fmt.Errorf("quorum not reached (ok=%d all=%d)", ok, all)
}
grainSyncCount.Inc()
return nil
}
func (p *SyncedPool) removeLocalGrain(id CartId) {
p.mu.Lock()
delete(p.local.grains, id)
p.mu.Unlock()
}
// getGrain returns a local or remote grain; if absent, attempts ownership.
func (p *SyncedPool) getGrain(id CartId) (Grain, error) {
p.mu.RLock()
localGrain, isLocal := p.local.grains[id]
remoteGrain, isRemote := p.remoteIndex[id]
p.mu.RUnlock()
if isLocal && localGrain != nil {
return localGrain, nil
}
if isRemote {
remoteLookupCount.Inc()
return remoteGrain, nil
}
// Attempt to claim ownership (async semantics preserved)
go p.RequestOwnership(id)
// Create local grain (lazy spawn) - may be rolled back by quorum failure
grain, err := p.local.GetGrain(id)
if err != nil {
return nil, err
}
return grain, nil
}
// Process applies mutation(s) to a grain (local or remote).
func (p *SyncedPool) Process(id CartId, messages ...Message) (*FrameWithPayload, error) {
grain, err := p.getGrain(id)
if err != nil {
return nil, err
}
var res *FrameWithPayload
for _, m := range messages {
res, err = grain.HandleMessage(&m, false)
if err != nil {
return nil, err
}
}
return res, nil
}
// Get returns current state of a grain (local or remote).
func (p *SyncedPool) Get(id CartId) (*FrameWithPayload, error) {
grain, err := p.getGrain(id)
if err != nil {
return nil, err
}
return grain.GetCurrentState()
}
// Close notifies remotes this host is terminating.
func (p *SyncedPool) Close() {
p.mu.RLock()
remotes := make([]*RemoteHostGRPC, 0, len(p.remoteHosts))
for _, r := range p.remoteHosts {
remotes = append(remotes, r)
}
p.mu.RUnlock()
for _, r := range remotes {
go func(rh *RemoteHostGRPC) {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
_, err := rh.ControlClient.Closing(ctx, &proto.ClosingNotice{Host: p.Hostname})
cancel()
if err != nil {
log.Printf("Close notify to %s failed: %v", rh.Host, err)
}
}(r)
}
}
-8
View File
@@ -1,8 +0,0 @@
/*
Legacy TCP networking (GenericListener / Frame protocol) has been removed
as part of the gRPC migration. This file intentionally contains no tests.
Keeping an empty Go file (with a package declaration) ensures the old
tcp-connection test target no longer runs without causing build issues.
*/
package main