vbatts/microservices-demo
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Making cartservice more reliable

Browse source 
1. Making sure we re-create redis connection upon disconnect
2. Fixed local cart store implementation to handle updates (useful for testing w/o redis)
3. Fixed windows scripts to work against redis correctly
This commit is contained in:
Simon Zeltser 2018-07-02 13:26:37 -07:00
parent 11c208a9f4
commit d457f7ec28
6 changed files with 179 additions and 103 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -4,3 +4,5 @@ pkg/
*.pyc
*.swp
*~
.vs/slnx.sqlite
.vs/microservices-demo/v15/.suo

43
src/cartservice/Program.cs
View file

@ -33,26 +33,36 @@ namespace cartservice
{
// Run the server in a separate thread and make the main thread busy waiting.
// The busy wait is because when we run in a container, we can't use techniques such as waiting on user input (Console.Readline())
Task.Run(async () =>
Task serverTask = Task.Run(async () =>
{
Console.WriteLine($"Trying to start a grpc server at {host}:{port}");
Server server = new Server
try
{
Services = { Hipstershop.CartService.BindService(new CartServiceImpl(cartStore)) },
Ports = { new ServerPort(host, port, ServerCredentials.Insecure) }
};
Console.WriteLine($"Trying to start a grpc server at {host}:{port}");
Server server = new Server
{
Services = { Hipstershop.CartService.BindService(new CartServiceImpl(cartStore)) },
Ports = { new ServerPort(host, port, ServerCredentials.Insecure) }
};
Console.WriteLine($"Cart server is listening at {host}:{port}");
server.Start();
Console.WriteLine($"Cart server is listening at {host}:{port}");
server.Start();
await cartStore.InitializeAsync();
await cartStore.InitializeAsync();
Console.WriteLine("Initialization completed");
// Keep the server up and running
while(true)
{
Thread.Sleep(TimeSpan.FromMinutes(10));
}
}
catch (Exception ex)
{
Console.WriteLine(ex);
}
});
// Busy wait to keep the process alive
while(true)
{
Thread.Sleep(TimeSpan.FromMinutes(10));
}
return Task.WaitAny(new[] { serverTask });
}
static void Main(string[] args)
@ -69,6 +79,8 @@ namespace cartservice
Parser.Default.ParseArguments<ServerOptions>(args).MapResult(
(ServerOptions options) =>
{
Console.WriteLine($"Started as process with id {System.Diagnostics.Process.GetCurrentProcess().Id}");
// Set hostname/ip address
string hostname = options.Host;
if (string.IsNullOrEmpty(hostname))
@ -106,7 +118,10 @@ namespace cartservice
// Redis was specified via command line or environment variable
if (!string.IsNullOrEmpty(redis))
{
// If you want to start cart store using local cache in process, you can replace the following line with this:
// cartStore = new LocalCartStore();
cartStore = new RedisCartStore(redis);
return StartServer(hostname, port, cartStore);
}
else

19
src/cartservice/cartstore/LocalCartStore.cs
View file

@ -1,6 +1,7 @@
using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;
using System.Linq;
using cartservice.interfaces;
using Hipstershop;
@ -10,6 +11,7 @@ namespace cartservice.cartstore
{
// Maps between user and their cart
private ConcurrentDictionary<string, Hipstershop.Cart> userCartItems = new ConcurrentDictionary<string, Hipstershop.Cart>();
private readonly Hipstershop.Cart emptyCart = new Hipstershop.Cart();
public Task InitializeAsync()
{
@ -29,8 +31,17 @@ namespace cartservice.cartstore
userCartItems.AddOrUpdate(userId, newCart,
(k, exVal) =>
{
// Currently we assume that we only add to the cart
exVal.Items.Add(new Hipstershop.CartItem { ProductId = productId, Quantity = quantity });
// If the item exists, we update its quantity
var existingItem = exVal.Items.SingleOrDefault(item => item.ProductId == productId);
if (existingItem != null)
{
existingItem.Quantity += quantity;
}
else
{
exVal.Items.Add(new Hipstershop.CartItem { ProductId = productId, Quantity = quantity });
}
return exVal;
});
@ -40,7 +51,7 @@ namespace cartservice.cartstore
public Task EmptyCartAsync(string userId)
{
Console.WriteLine($"EmptyCartAsync called with userId={userId}");
userCartItems[userId] = new Hipstershop.Cart();
userCartItems[userId] = emptyCart;
return Task.CompletedTask;
}
@ -52,7 +63,9 @@ namespace cartservice.cartstore
if (!userCartItems.TryGetValue(userId, out cart))
{
Console.WriteLine($"No carts for user {userId}");
return Task.FromResult(emptyCart);
}
return Task.FromResult(cart);
}
}

191
src/cartservice/cartstore/RedisCartStore.cs
View file

@ -1,9 +1,11 @@
using System;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using cartservice.interfaces;
using Google.Protobuf;
using Grpc.Core;
using Hipstershop;
using StackExchange.Redis;
@ -12,137 +14,174 @@ namespace cartservice.cartstore
public class RedisCartStore : ICartStore
{
private const string CART_FIELD_NAME = "cart";
private const int REDIS_RETRY_NUM = 5;
private static ConnectionMultiplexer redis;
private volatile ConnectionMultiplexer redis;
private volatile bool isRedisConnectionOpened = false;
private readonly object locker = new object();
private readonly byte[] emptyCartBytes;
<<<<<<< HEAD
private readonly string connectionString;
private readonly string redisAddr;
||||||| merged common ancestors
=======
private readonly string connectionString;
>>>>>>> origin
private readonly ConfigurationOptions redisConnectionOptions;
public RedisCartStore(string redisAddress)
{
// Serialize empty cart into byte array.
var cart = new Hipstershop.Cart();
emptyCartBytes = cart.ToByteArray();
this.redisAddr = redisAddress;
connectionString = $"{redisAddress},ssl=false,allowAdmin=true,connectRetry=5";
Console.WriteLine($"Going to use Redis cache at this address: {connectionString}");
redisConnectionOptions = ConfigurationOptions.Parse(connectionString);
// Try to reconnect if first retry failed (up to 5 times with exponential backoff)
redisConnectionOptions.ConnectRetry = REDIS_RETRY_NUM;
redisConnectionOptions.ReconnectRetryPolicy = new ExponentialRetry(100);
redisConnectionOptions.KeepAlive = 180;
}
<<<<<<< HEAD
public Task InitializeAsync()
{
||||||| merged common ancestors
string connectionString = $"{redisAddress},ssl=false,allowAdmin=true";
=======
connectionString = $"{redisAddress},ssl=false,allowAdmin=true";
Console.WriteLine($"Going to use Redis cache at this address: {connectionString}");
EnsureRedisConnected();
return Task.CompletedTask;
}
public async Task InitializeAsync()
private void EnsureRedisConnected()
{
>>>>>>> origin
Console.WriteLine("Connecting to Redis: " + connectionString);
<<<<<<< HEAD
if (isRedisConnectionOpened)
{
return;
}
redis = ConnectionMultiplexer.Connect(connectionString);
Console.WriteLine("Connected successfully to Redis");
// Connection is closed or failed - open a new one but only at the first thread
lock (locker)
{
if (isRedisConnectionOpened)
{
return;
}
return Task.CompletedTask;
||||||| merged common ancestors
redis = ConnectionMultiplexer.Connect(connectionString);
=======
redis = await ConnectionMultiplexer.ConnectAsync(connectionString, Console.Out);
Console.WriteLine("Connected successfully to Redis");
>>>>>>> origin
Console.WriteLine("Connecting to Redis: " + connectionString);
redis = ConnectionMultiplexer.Connect(redisConnectionOptions);
if (redis == null || !redis.IsConnected)
{
Console.WriteLine("Wasn't able to connect to redis");
// We weren't able to connect to redis despite 5 retries with exponential backoff
throw new ApplicationException("Wasn't able to connect to redis");
}
Console.WriteLine("Successfully connected to Redis");
var cache = redis.GetDatabase();
Console.WriteLine("Performing small test");
cache.StringSet("cart", "OK" );
object res = cache.StringGet("cart");
Console.WriteLine($"Small test result: {res}");
redis.InternalError += (o, e) => { Console.WriteLine(e.Exception); };
redis.ConnectionRestored += (o, e) =>
{
isRedisConnectionOpened = true;
Console.WriteLine("Connection to redis was retored successfully");
};
redis.ConnectionFailed += (o, e) =>
{
Console.WriteLine("Connection failed. Disposing the object");
isRedisConnectionOpened = false;
};
isRedisConnectionOpened = true;
}
}
public async Task AddItemAsync(string userId, string productId, int quantity)
{
Console.WriteLine($"AddItemAsync called with userId={userId}, productId={productId}, quantity={quantity}");
var db = redis.GetDatabase();
// Access the cart from the cache
var value = await db.HashGetAsync(userId, CART_FIELD_NAME);
try
{
EnsureRedisConnected();
var db = redis.GetDatabase();
// Access the cart from the cache
var value = await db.HashGetAsync(userId, CART_FIELD_NAME);
Hipstershop.Cart cart;
if (value.IsNull)
{
cart = new Hipstershop.Cart();
cart.UserId = userId;
cart.Items.Add(new Hipstershop.CartItem { ProductId = productId, Quantity = quantity });
}
else
{
cart = Hipstershop.Cart.Parser.ParseFrom(value);
var existingItem = cart.Items.SingleOrDefault(i => i.ProductId == productId);
if (existingItem == null)
Hipstershop.Cart cart;
if (value.IsNull)
{
cart = new Hipstershop.Cart();
cart.UserId = userId;
cart.Items.Add(new Hipstershop.CartItem { ProductId = productId, Quantity = quantity });
}
else
{
existingItem.Quantity += quantity;
cart = Hipstershop.Cart.Parser.ParseFrom(value);
var existingItem = cart.Items.SingleOrDefault(i => i.ProductId == productId);
if (existingItem == null)
{
cart.Items.Add(new Hipstershop.CartItem { ProductId = productId, Quantity = quantity });
}
else
{
existingItem.Quantity += quantity;
}
}
}
await db.HashSetAsync(userId, new[]{ new HashEntry(CART_FIELD_NAME, cart.ToByteArray()) });
await db.HashSetAsync(userId, new[]{ new HashEntry(CART_FIELD_NAME, cart.ToByteArray()) });
}
catch (Exception ex)
{
throw new RpcException(new Status(StatusCode.FailedPrecondition, $"Can't access cart storage. {ex}"));
}
}
public async Task EmptyCartAsync(string userId)
{
Console.WriteLine($"EmptyCartAsync called with userId={userId}");
var db = redis.GetDatabase();
try
{
EnsureRedisConnected();
var db = redis.GetDatabase();
// Update the cache with empty cart for given user
await db.HashSetAsync(userId, new[] { new HashEntry(CART_FIELD_NAME, emptyCartBytes) });
// Update the cache with empty cart for given user
await db.HashSetAsync(userId, new[] { new HashEntry(CART_FIELD_NAME, emptyCartBytes) });
}
catch (Exception ex)
{
throw new RpcException(new Status(StatusCode.FailedPrecondition, $"Can't access cart storage. {ex}"));
}
}
public async Task<Hipstershop.Cart> GetCartAsync(string userId)
{
Console.WriteLine($"GetCartAsync called with userId={userId}");
<<<<<<< HEAD
try
{
var db = redis.GetDatabase();
// Access the cart from the cache
||||||| merged common ancestors
var db = redis.GetDatabase();
=======
try
{
var db = redis.GetDatabase();
>>>>>>> origin
<<<<<<< HEAD
var value = await db.HashGetAsync(userId, CART_FIELD_NAME);
||||||| merged common ancestors
// Access the cart from the cache
var value = await db.HashGetAsync(userId, CART_FIELD_NAME);
=======
try
{
EnsureRedisConnected();
var db = redis.GetDatabase();
// Access the cart from the cache
var value = await db.HashGetAsync(userId, CART_FIELD_NAME);
>>>>>>> origin
if (!value.IsNull)
{
return Hipstershop.Cart.Parser.ParseFrom(value);
}
}
catch (Exception e)
{
Console.WriteLine(e);
}
// We decided to return empty cart in cases when user wasn't in the cache before
return new Hipstershop.Cart();
// We decided to return empty cart in cases when user wasn't in the cache before
return new Hipstershop.Cart();
}
catch (Exception ex)
{
throw new RpcException(new Status(StatusCode.FailedPrecondition, $"Can't access cart storage. {ex}"));
}
}
}
}

22
src/cartservice/scripts/docker_setup.bat
View file

@ -9,12 +9,13 @@ GOTO End1
:local
set REDIS_PORT=6379
set REDIS_ADDR=localhost:%REDIS_PORT%
set LISTEN_ADDR=0.0.0.0
set LISTEN_ADDR=localhost
set PORT=7070
set GRPC_TRACE=all
echo running redis emulator locally on a separate window
taskkill /f /im "redis-server.exe"
start redis-server
start redis-server "C:\ProgramData\chocolatey\lib\redis-64\redis.windows.conf"
echo running the cart service locally
dotnet build ..\.
@ -23,19 +24,24 @@ GOTO End1
:docker_local
set REDIS_PORT=6379
set REDIS_ADDR=redis:%REDIS_PORT%
set LISTEN_ADDR=0.0.0.0
rem set REDIS_ADDR=redis:%REDIS_PORT%
set LISTEN_ADDR=localhost
set PORT=7070
echo run docker container with redis
docker rm --force redis
start "" docker run -d --name=redis -p %REDIS_PORT%:%REDIS_PORT% redis
echo Forcing to remove redis cache so we always start the container from scratch
docker rm --force redis > nul 2>&1
echo Starting out redis container
docker run -d --name=redis redis > nul 2>&1
rem This assigns the output of ip4 addr of redis container into REDIS_ADDR
FOR /F "tokens=*" %%g IN ('docker inspect -f "{{ .NetworkSettings.Networks.bridge.IPAddress }}" redis') do (SET REDIS_ADDR=%%g)
echo addr=%REDIS_ADDR%
echo building container image for cart service
docker build -t cartservice ..\.
echo run container image for cart service
docker run -it --rm -e REDIS_ADDR=%REDIS_ADDR% -e LISTEN_ADDR=%LISTEN_ADDR% -e PORT=%PORT% -p %PORT%:%PORT% cartservice
docker run -it --name=cartservice --rm -e REDIS_ADDR=%REDIS_ADDR%:%REDIS_PORT% -e LISTEN_ADDR=%LISTEN_ADDR% -e PORT=%PORT% -p %PORT%:%PORT% cartservice
GOTO End1

5
tests/cartservice/CartServiceTests.cs
View file

@ -29,15 +29,16 @@ namespace cartservice
{
UserId = userId,
};
var cart = await client.GetCartAsync(request);
Assert.NotNull(cart);
// All grpc objects implement IEquitable, so we can compare equality with by-value semantics
Assert.Equal(new Cart(), cart);
}
[Fact]
public async Task AddItem_ItemExists_Udpated()
public async Task AddItem_ItemExists_Updated()
{
string userId = Guid.NewGuid().ToString();